tests/populate_jobs.rs
//! Inject-assert (the "jobview gap"): a server-side **workspace populate**
//! (register-eager → clone every member → consume them into the code room) MUST
//! surface as JobRecords in the ledger, so it is visible in the Jobs panel +
//! `nornir jobs`.
//!
//! Since Mímir v2 the build half of a populate IS the consumer: after the
//! clones, every member is consumed at its tip by the nightly driver, as a
//! `workspace_republish` child of the populate parent. So the parent is `done`
//! only on a box that has the driver, and without one it ends `failed` with
//! the state that stopped it (`driver_not_available`) — the members are cloned
//! either way. Both are asserted here.
//!
//! This obeys the test-matrix LAWS:
//! * LAW 4 (END-TO-END = THROUGH THE SERVER): the populate is driven via the
//! real `Workspaces.Register` gRPC against a running `nornir-server`, and the
//! jobs are read BACK through the server via `Viz.Jobs` (`fetch_jobs`), not a
//! local ledger read.
//! * LAW 2 (RAGNARÖK + its inverse): a populate WITH members asserts `> 0` jobs
//! covering the clone + consume phases, with terminal status; the inverse is
//! the SAME `Viz.Jobs` oracle reading `0` jobs for the workspace BEFORE any
//! populate has run (empty → 0). A surface that can't go to 0 on no-input is
//! asserting nothing.
//!
//! Real input: two `file://`-cloneable Rust-crate fixtures (no network), and the
//! driver built from `crates/mimir-driver` on first use (as
//! `crates/mimir-ingest/tests/ingest/real.rs` does).
//!
//! Run: cargo test --features "viz,server" --test populate_jobs -- --test-threads=1
#![cfg(all(feature = "viz", feature = "server"))]
use std::net::TcpListener;
use std::path::{Path, PathBuf};
use std::process::{Child, Command};
use std::time::{Duration, Instant};
use nornir::jobs::{JobRecord, kind, status};
use uuid::Uuid;
fn free_port() -> u16 {
TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
/// A minimal committed Rust-crate fixture repo a pure-gix `file://` clone can fetch.
fn fixture_crate(dir: &Path, name: &str) {
std::fs::create_dir_all(dir.join("src")).unwrap();
std::fs::write(
dir.join("Cargo.toml"),
format!("[package]\nname = \"{name}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n\n[dependencies]\n"),
)
.unwrap();
std::fs::write(
dir.join("src").join("lib.rs"),
format!("//! {name} fixture crate\npub fn {name}_answer() -> i32 {{ 42 }}\n"),
)
.unwrap();
nornir::gitio::init(dir).unwrap();
nornir::gitio::commit_all(dir, "init fixture").unwrap();
}
fn file_url(dir: &Path) -> String {
format!("file://{}", dir.display())
}
/// An ephemeral `nornir-server` with a FRESH EMPTY registry (no workspaces served).
struct Server {
child: Child,
root: PathBuf,
endpoint: String,
token: String,
}
impl Drop for Server {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
let _ = std::fs::remove_dir_all(&self.root);
}
}
/// `cargo`'s own variables and the wrappers the outer `cargo test` runs under:
/// neither the driver build nor the server's driver runs may inherit them.
fn steers_cargo(k: &str) -> bool {
(k.starts_with("CARGO_") && k != "CARGO_HOME") || matches!(k, "RUSTFLAGS" | "RUSTC_WRAPPER" | "RUSTC_WORKSPACE_WRAPPER" | "RUSTUP_TOOLCHAIN" | "RUSTC")
}
/// Build mimir-driver (its own workspace, pinned nightly) once; `cargo-mimir-dump`.
fn built_driver() -> PathBuf {
let dir = PathBuf::from(concat!(env!("CARGO_MANIFEST_DIR"), "/crates/mimir-driver"));
let mut c = Command::new("cargo");
for (k, _) in std::env::vars() {
if steers_cargo(&k) {
c.env_remove(k);
}
}
let st = c.current_dir(&dir).args(["build", "--release", "--bins"]).status().expect("run cargo for mimir-driver");
assert!(st.success(), "building mimir-driver failed");
dir.join("target/release/cargo-mimir-dump")
}
/// A server on a box WITH the nightly driver: its populate consumes.
fn spawn_empty_server() -> Server {
spawn_server(Some(&built_driver()))
}
/// `driver: None` = a box without the driver (nothing named, nothing to find
/// under the scratch home or beside the server binary).
fn spawn_server(driver: Option<&Path>) -> Server {
let token = "populate-jobs-token-0123456789abcdef";
let port = free_port();
let endpoint = format!("http://127.0.0.1:{port}");
let root = std::env::temp_dir().join(format!(
"nornir-populate-jobs-{}-{}",
std::process::id(),
Uuid::new_v4()
));
let _ = std::fs::remove_dir_all(&root);
let server_home = root.join("home");
std::fs::create_dir_all(server_home.join(".nornir").join("workspaces")).unwrap();
// the server's HOME is scratch; rustup and cargo keep the real ones, or the
// driver's `cargo +nightly` would find no toolchain
let home = PathBuf::from(std::env::var_os("HOME").expect("HOME"));
let rustup_home = std::env::var_os("RUSTUP_HOME").map(PathBuf::from).unwrap_or_else(|| home.join(".rustup"));
let cargo_home = std::env::var_os("CARGO_HOME").map(PathBuf::from).unwrap_or_else(|| home.join(".cargo"));
let mut cmd = Command::new(env!("CARGO_BIN_EXE_nornir-server"));
cmd.env("NORNIR_SERVER_ADDR", format!("127.0.0.1:{port}"))
.env("NORNIR_SERVER_TOKEN", token)
.env("HOME", &server_home)
.env("RUSTUP_HOME", rustup_home)
.env("CARGO_HOME", cargo_home)
// Long poll so the SWEEP loop never races our register — we assert the
// register-eager populate itself produces the jobs, not a later poll tick.
.env("NORNIR_POLL", "3600s")
.env_remove("NORNIR_MIMIR_DRIVER")
.env_remove("NORNIR_MIMIR_DRIVER_HOST")
.env_remove("NORNIR_MIMIR_TOOLCHAIN");
for (k, _) in std::env::vars() {
if steers_cargo(&k) {
cmd.env_remove(k);
}
}
match driver {
Some(d) => {
cmd.env("NORNIR_MIMIR_DRIVER", d);
}
// the search also covers PATH: leave nothing there that could hold one
None => {
cmd.env("PATH", "/usr/bin:/bin");
}
}
let child = cmd
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("spawn nornir-server");
let srv = Server {
child,
root,
endpoint: endpoint.clone(),
token: token.to_string(),
};
let deadline = Instant::now() + Duration::from_secs(45);
loop {
if nornir::viz::remote::list_workspaces(&endpoint, token).is_ok() {
break;
}
assert!(Instant::now() < deadline, "empty server never came up");
std::thread::sleep(Duration::from_millis(200));
}
srv
}
/// Write a monitored-workspace descriptor with two `file://`-cloneable members.
fn write_descriptor(dir: &Path, ws_name: &str) -> PathBuf {
let src_alpha = dir.join("src_alpha");
let src_beta = dir.join("src_beta");
fixture_crate(&src_alpha, "alpha");
fixture_crate(&src_beta, "beta");
let desc = dir.join("nornir-workspace.toml");
std::fs::write(
&desc,
format!(
"[workspace]\nname = \"{ws_name}\"\n\n\
[repos.alpha]\ngit = \"{}\"\n\n\
[repos.beta]\ngit = \"{}\"\n",
file_url(&src_alpha),
file_url(&src_beta),
),
)
.unwrap();
desc
}
/// Poll `Viz.Jobs` (through the server) until `pick` returns `Some`, or panic at
/// the deadline (generous: the populate compiles the members under the driver).
/// The eager populate runs on a `spawn_blocking` task AFTER register returns, so
/// the jobs land asynchronously — we poll the SERVER, not a local read.
fn poll_jobs_until<T>(srv: &Server, ws: &str, pick: impl Fn(&[JobRecord]) -> Option<T>) -> T {
let deadline = Instant::now() + Duration::from_secs(600);
let mut last = String::from("no answer yet");
loop {
match nornir::viz::remote::fetch_jobs(&srv.endpoint, &srv.token, ws, "", 0) {
Ok(rows) => {
if let Some(v) = pick(&rows) {
return v;
}
last = rows.iter().map(|r| format!("{} {} {} {}", r.kind, r.target, r.status, r.detail_json)).collect::<Vec<_>>().join(" | ");
}
Err(e) => last = format!("Viz.Jobs failed: {e:#}"),
}
assert!(
Instant::now() < deadline,
"populate jobs never appeared for `{ws}` via Viz.Jobs — last seen: {last}"
);
std::thread::sleep(Duration::from_millis(250));
}
}
/// The workspace's rows once its populate parent is over — `done` or `failed`:
/// a populate that failed must fail the assertion that reads it, not a timeout.
fn populate_is_over(rows: &[JobRecord], ws: &str) -> Option<Vec<JobRecord>> {
let ours: Vec<&JobRecord> = rows.iter().filter(|r| r.workspace == ws).collect();
ours.iter()
.any(|r| r.kind == kind::WORKSPACE_POPULATE && (r.status == status::DONE || r.status == status::FAILED))
.then(|| ours.into_iter().cloned().collect())
}
/// A box WITHOUT the driver: the members are still cloned (two done children),
/// and the populate parent ends `failed` naming the state that stopped the
/// consume — never `done` as if a code room had been built.
#[test]
fn a_populate_without_the_driver_clones_the_members_and_fails_by_state() {
let srv = spawn_server(None);
let ws = "nodriver";
let fixtures = srv.root.join("fixtures");
std::fs::create_dir_all(&fixtures).unwrap();
let desc = write_descriptor(&fixtures, ws);
let (_name, _mode, members) =
nornir::viz::remote::register_workspace(&srv.endpoint, &srv.token, ws, &desc.to_string_lossy(), "monitored", "3600s").expect("register must succeed");
assert_eq!(members, 2);
let rows = poll_jobs_until(&srv, ws, |rows| populate_is_over(rows, ws));
let parent = rows.iter().find(|r| r.kind == kind::WORKSPACE_POPULATE).expect("the populate parent");
assert_eq!(parent.status, status::FAILED, "no driver, no code room: {parent:?}");
let detail: serde_json::Value = serde_json::from_str(&parent.detail_json).expect("the parent's detail is JSON");
assert_eq!(detail["fetched"], 2, "{detail}");
assert_eq!(detail["consumed"]["refusal"]["state"], "driver_not_available", "{detail}");
let clones: Vec<&JobRecord> = rows.iter().filter(|r| r.kind == kind::WORKSPACE_CLONE).collect();
assert_eq!(clones.len(), 2, "{rows:?}");
assert!(clones.iter().all(|c| c.status == status::DONE && c.parent_id.as_deref() == Some(parent.job_id.as_str())), "{clones:?}");
assert!(rows.iter().all(|r| r.kind != kind::WORKSPACE_REPUBLISH), "no consume run was started: {rows:?}");
}
#[test]
fn register_eager_populate_emits_clone_jobs_and_consumes_the_members() {
let srv = spawn_empty_server();
let ws = "populatejobs";
// Fixtures + descriptor under the server's root (shared fs, server-readable).
let fixtures = srv.root.join("fixtures");
std::fs::create_dir_all(&fixtures).unwrap();
let desc = write_descriptor(&fixtures, ws);
// ── LAW 2 inverse (empty → 0): the SAME Viz.Jobs oracle reads 0 jobs for the
// workspace BEFORE any populate has run. (An unserved workspace's Viz.Jobs
// falls back to the empty/global ledger; the key point is there are 0 jobs
// for `populatejobs` until the populate runs.)
let before =
nornir::viz::remote::fetch_jobs(&srv.endpoint, &srv.token, ws, "", 0).unwrap_or_default();
let before_ws: Vec<_> = before.iter().filter(|r| r.workspace == ws).collect();
assert!(
before_ws.is_empty(),
"RAGNARÖK inverse: a never-populated workspace must have 0 jobs, got {before_ws:?}"
);
// ── Drive the populate THROUGH THE SERVER (LAW 4): register-eager clones both
// members and builds the warehouse. ──
let (name, mode, members) = nornir::viz::remote::register_workspace(
&srv.endpoint,
&srv.token,
ws,
&desc.to_string_lossy(),
"monitored",
"3600s",
)
.expect("runtime register (eager populate) must succeed");
assert_eq!(name, ws);
assert_eq!(mode, "monitored");
assert_eq!(
members, 2,
"eager populate cloned both members from the descriptor"
);
// ── LAW 2 (RAGNARÖK) + HIERARCHY: the populate produced a PARENT
// `workspace_populate` job plus STREAMED children — one `workspace_clone`
// per member, then the consume run — read BACK through the server
// (Viz.Jobs), each terminal. The parent finishing last means polling for a
// terminal parent guarantees every child has landed too. ──
let rows = poll_jobs_until(&srv, ws, |rows| populate_is_over(rows, ws));
// The PARENT populate job (top-level, no parent_id).
let parent = rows
.iter()
.find(|r| r.kind == kind::WORKSPACE_POPULATE)
.expect("the populate umbrella is a tracked parent job");
assert_eq!(
parent.status,
status::DONE,
"populate parent finished done: {parent:?}"
);
assert_eq!(parent.parent_id, None, "the populate parent is top-level");
let detail: serde_json::Value = serde_json::from_str(&parent.detail_json).expect("the parent's detail is JSON");
assert_eq!(detail["fetched"], 2, "populate parent rolls up the member count: {detail}");
// Children: the two member clones and the consume run hang off the parent.
let children: Vec<&JobRecord> = rows
.iter()
.filter(|r| r.parent_id.as_deref() == Some(parent.job_id.as_str()))
.collect();
// One `workspace_clone` CHILD per member (alpha + beta), each terminal+done.
let clones: Vec<&&JobRecord> = children
.iter()
.filter(|r| r.kind == kind::WORKSPACE_CLONE)
.collect();
assert_eq!(
clones.len(),
2,
"one clone child PER MEMBER streamed in: {clones:?}"
);
for c in &clones {
assert_eq!(c.status, status::DONE, "member clone finished done: {c:?}");
assert_eq!(c.parent_id.as_deref(), Some(parent.job_id.as_str()));
}
let mut targets: Vec<&str> = clones.iter().map(|c| c.target.as_str()).collect();
targets.sort();
assert_eq!(
targets,
["alpha", "beta"],
"a clone child per descriptor member"
);
// The build half IS the consumer (Mímir v2): ONE `workspace_republish` child
// under the parent, done, and the parent's detail says what it consumed —
// each member at its tip, one commit each, both caught up.
let consumes: Vec<&&JobRecord> = children.iter().filter(|r| r.kind == kind::WORKSPACE_REPUBLISH).collect();
assert_eq!(consumes.len(), 1, "one consume run under the populate: {children:?}");
assert_eq!(consumes[0].status, status::DONE, "the consume run finished done: {:?}", consumes[0]);
assert_eq!(children.len(), 3, "two clones and the consume run, nothing else: {children:?}");
assert_eq!(detail["consumed"]["commits"], 2, "one commit per member, from its tip: {detail}");
let members: Vec<&str> = detail["consumed"]["members"].as_array().expect("consumed.members").iter().filter_map(|m| m.as_str()).collect();
assert_eq!(members.len(), 2, "{detail}");
for name in ["alpha", "beta"] {
assert!(
members.iter().any(|m| m.starts_with(name) && m.contains("caught_up")),
"`{name}` was consumed and is caught up: {members:?}"
);
}
// … one `knowledge_member` row per consumed commit, under the consume run.
let commits: Vec<&JobRecord> = rows
.iter()
.filter(|r| r.kind == kind::KNOWLEDGE_MEMBER && r.parent_id.as_deref() == Some(consumes[0].job_id.as_str()))
.collect();
assert_eq!(commits.len(), 2, "one row per (member, commit): {commits:?}");
assert!(commits.iter().all(|r| r.status == status::DONE), "{commits:?}");
// ── `--kind workspace_clone` scoping through the server returns the per-member
// clone children — exactly what `nornir jobs --kind workspace_clone --server`
// would. ──
let only_clone =
nornir::viz::remote::fetch_jobs(&srv.endpoint, &srv.token, ws, kind::WORKSPACE_CLONE, 0)
.expect("Viz.Jobs --kind workspace_clone");
let only_clone_ws: Vec<_> = only_clone.iter().filter(|r| r.workspace == ws).collect();
assert_eq!(
only_clone_ws.len(),
2,
"two clone children for the two members"
);
assert!(
only_clone_ws
.iter()
.all(|r| r.kind == kind::WORKSPACE_CLONE)
);
}
/// **Nuke → repopulate** — the viz "🗑 Kill workspace" (`--purge`) then "⬇ Populate"
/// cycle, end-to-end THROUGH THE SERVER (LAW 4). Asserts:
/// 1. a populate builds the on-disk `builds/` warehouse and registers the row;
/// 2. `Workspaces.Remove { purge: true }` drops the registry row AND
/// `remove_dir_all`s the workspace's `builds/` (the gap the `--purge` fix closed —
/// unregister-only used to leak the on-disk warehouse);
/// 3. re-registering REPOPULATES from scratch — a FRESH populate job (distinct
/// `job_id` from the first, since the jobs ledger is NOT purged) lands `done`
/// and the warehouse is rebuilt.
#[test]
fn nuke_with_purge_removes_builds_then_repopulate_reemits_a_fresh_populate() {
let srv = spawn_empty_server();
let ws = "nukerepop";
let fixtures = srv.root.join("fixtures");
std::fs::create_dir_all(&fixtures).unwrap();
let desc = write_descriptor(&fixtures, ws);
// The per-workspace on-disk warehouse: `<registry root>/<ws>/builds/`.
let builds = srv
.root
.join("home")
.join(".nornir")
.join("workspaces")
.join(ws)
.join("builds");
// ── 1) Populate → a done populate parent + the builds/ warehouse on disk. ──
let (_n, _m, members) = nornir::viz::remote::register_workspace(
&srv.endpoint,
&srv.token,
ws,
&desc.to_string_lossy(),
"monitored",
"3600s",
)
.expect("first populate must succeed");
assert_eq!(members, 2, "first populate cloned both members");
let first_parent_id = poll_jobs_until(&srv, ws, |rows| {
rows.iter()
.find(|r| {
r.workspace == ws && r.kind == kind::WORKSPACE_POPULATE && r.status == status::DONE
})
.map(|r| r.job_id.clone())
});
assert!(
nornir::viz::remote::list_workspaces(&srv.endpoint, &srv.token)
.unwrap()
.iter()
.any(|w| w == ws),
"workspace is registered after the populate"
);
assert!(
builds.exists(),
"populate built the on-disk warehouse at {}",
builds.display()
);
// ── 2) NUKE with purge → registry row gone AND builds/ removed. ──
nornir::viz::remote::remove_workspace(&srv.endpoint, &srv.token, ws, true)
.expect("Workspaces.Remove { purge: true } must succeed");
assert!(
!nornir::viz::remote::list_workspaces(&srv.endpoint, &srv.token)
.unwrap()
.iter()
.any(|w| w == ws),
"nuke dropped the registry row"
);
assert!(
!builds.exists(),
"--purge removed the on-disk builds/ warehouse at {}",
builds.display()
);
// ── 3) REPOPULATE → re-register re-clones + rebuilds; a FRESH populate job
// (distinct id from the first) lands done, and the warehouse is back. ──
let (_n2, _m2, members2) = nornir::viz::remote::register_workspace(
&srv.endpoint,
&srv.token,
ws,
&desc.to_string_lossy(),
"monitored",
"3600s",
)
.expect("repopulate after nuke must succeed");
assert_eq!(members2, 2, "repopulate cloned both members again");
let second_parent_id = poll_jobs_until(&srv, ws, |rows| {
rows.iter()
.find(|r| {
r.workspace == ws
&& r.kind == kind::WORKSPACE_POPULATE
&& r.status == status::DONE
&& r.job_id != first_parent_id
})
.map(|r| r.job_id.clone())
});
assert_ne!(
second_parent_id, first_parent_id,
"repopulate emitted a NEW populate job, not the stale pre-nuke one"
);
assert!(
builds.exists(),
"repopulate rebuilt the on-disk warehouse at {}",
builds.display()
);
}