tests/register_across_processes.rs
//! **THE PROOF FOR H6: a SEPARATE PROCESS reads the live state of a collector another
//! process is nursing.**
//!
//! Everything else about the register can be asserted in one process, and none of it is
//! evidence for the property that matters. `live_facet` was readable in-process all along;
//! what could not be done was reading it from outside, and that is the whole difference
//! between a supervisor and a thread pool. So this file spawns **two real child processes**
//! of the built `korp-collectors` binary:
//!
//! ```text
//! child A: korp-collectors serve --data-dir D ← holds redb's exclusive lock
//! child B: korp-collectors status --data-dir D ← different pid, no lock, reads
//! ```
//!
//! and asserts that B prints A's collectors, their phases **and their facets**, while A is
//! still running. If it cannot, the hole is not closed however tidy the code is.
//!
//! # What makes each assertion evidence rather than decoration
//!
//! * **The pids are compared.** A status verb that happened to run in-process would pass
//! every other assertion here.
//! * **The reader runs WHILE the writer holds the lock.** redb takes
//! `flock(LOCK_EX | LOCK_NB)`; five `mimir analyze` lenses died on exactly this. A read
//! that only works after the service stops is not a read of live state.
//! * **The facet is asserted, not just the phase.** A phase-only register would satisfy a
//! test that checked `phase != null` and still leave every out-of-process substrate
//! without the collector's own provenance.
//! * **Exit codes are asserted unpiped**, and the child's stdout is captured rather than
//! piped to a shell: a command piped to `tail` reports exit 0, which is how five green
//! verdicts were taken off failing commands in one day.
//! * **The red probe kills the writer** and asserts the same row then reads `orphaned`
//! with exit 3. Without it, a register that never checked liveness would pass every
//! green above.
#![cfg(all(unix, feature = "service"))]
use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
mod common;
use common::emit_for;
/// The built binary under test. Cargo sets this for a bin target of this crate, so the test
/// cannot accidentally exercise a stale copy on `$PATH`.
const BIN: &str = env!("CARGO_BIN_EXE_korp-collectors");
/// A roster the child can actually run without a network: `dcat` pointed at a port nothing
/// listens on. It goes Offline → work → `Failed`, restarting, and that is fine — the
/// property under test is that its live state is visible from outside, not that it succeeds.
fn roster(data_dir: &Path) -> String {
serde_json::json!({
"data_dir": data_dir,
"heartbeat_ms": 300,
"backoff_min_ms": 200,
"backoff_max_ms": 400,
"substrate": "nursery-under-test",
"unit": "collector-nursery.service",
"collectors": [
{ "name": "dcat-ap-se", "config": { "url": "http://127.0.0.1:1/dcat.rdf" } },
{ "name": "pxweb-jsonstat2", "enabled": false }
]
})
.to_string()
}
/// Spawn `serve` as a real child process. Its stderr is inherited so a failure is visible
/// in the test log rather than swallowed.
fn spawn_serve(data_dir: &Path) -> Child {
Command::new(BIN)
.arg("serve")
.arg("--data-dir")
.arg(data_dir)
.env("COLLECTOR_NURSERY_JSON", roster(data_dir))
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn the collector service as a child process")
}
/// Run `status` as a real child process and return `(exit_code, stdout)`.
///
/// The exit code comes from `ExitStatus::code()`, not from the shape of the output — a
/// wrapper that ends in `echo` gives exit 0 over a FAILED, and this test's whole value is
/// that it does not do that.
fn run_status(data_dir: &Path, json: bool) -> (i32, String) {
let mut cmd = Command::new(BIN);
cmd.arg("status").arg("--data-dir").arg(data_dir);
if json {
cmd.arg("--json");
}
let out = cmd
.output()
.expect("run the status verb as a child process");
(
out.status.code().unwrap_or(-1),
String::from_utf8_lossy(&out.stdout).to_string(),
)
}
/// Wait until the register holds a row for `collector` whose phase has moved past the
/// startup publish, or give up. Returns the parsed status JSON.
fn wait_for_row(data_dir: &Path, collector: &str, deadline: Duration) -> serde_json::Value {
let t0 = Instant::now();
let mut last = String::new();
while t0.elapsed() < deadline {
let (_code, out) = run_status(data_dir, true);
last = out.clone();
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&out) {
let found = v["collectors"]
.as_array()
.map(|a| a.iter().any(|r| r["collector"] == collector))
.unwrap_or(false);
if found {
return v;
}
}
std::thread::sleep(Duration::from_millis(100));
}
panic!("no row for `{collector}` within {deadline:?}; last status output was:\n{last}");
}
fn kill_and_reap(mut child: Child) -> Option<i32> {
let _ = child.kill();
let st = child.wait().ok();
// Drain the pipes so a failure in the child is readable when this test fails.
if let Some(mut e) = child.stderr.take() {
let mut s = String::new();
let _ = e.read_to_string(&mut s);
if !s.is_empty() {
eprintln!("--- child stderr ---\n{s}\n--------------------");
}
}
st.and_then(|s| s.code())
}
/// A data directory **under the running account's home**, because that is the contract the
/// nursery enforces: a service's data path and its unit's `User=` must name the same
/// account, and the check is not weakened for a test's convenience.
///
/// `std::env::temp_dir()` is deliberately not used. It would be refused — correctly, and
/// the first version of this file was refused by it — and on these boxes `/tmp` is tmpfs,
/// which is RAM, which is never where a service's durable data belongs.
fn tmp() -> PathBuf {
static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
let home = std::env::var("HOME").expect(
"this test needs $HOME — the nursery checks the \
data path against the running account's home",
);
let d = PathBuf::from(home).join(".cache").join(format!(
"kc-register-xproc-{}-{}",
std::process::id(),
N.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
));
std::fs::create_dir_all(&d).unwrap();
d
}
/// ★★ **THE DECISIVE TEST.** One process nurses, a different process reads — phase, facet
/// and all — while the first is still alive and holding the store's exclusive lock.
#[test]
fn a_separate_process_reads_the_live_state_of_a_collector_this_one_does_not_nurse() {
let data = tmp();
let serve = spawn_serve(&data);
let serve_pid = serve.id();
let v = wait_for_row(&data, "dcat-ap-se", Duration::from_secs(30));
// ── the register is a FILE, written by the other process ─────────────────────
let register = data.join(korp_collectors::register::REGISTER_DB_FILE);
assert!(
register.is_file(),
"the register must exist on disk at {} — an in-memory answer proves nothing about \
another process",
register.display()
);
// ── the reader is a DIFFERENT process ───────────────────────────────────────
let reader_pid = v["reader_pid"].as_u64().expect("reader_pid") as u32;
assert_ne!(
reader_pid,
std::process::id(),
"the status verb ran in THIS process — that is not the property under test"
);
assert_ne!(
reader_pid, serve_pid,
"the reader and the nursery must be different processes"
);
let rows = v["collectors"].as_array().expect("collectors array");
let dcat = rows
.iter()
.find(|r| r["collector"] == "dcat-ap-se")
.expect("dcat row");
// ── the row was written by the NURSERY, and says so ─────────────────────────
assert_eq!(
dcat["writer_pid"].as_u64().unwrap() as u32,
serve_pid,
"the row must name the nursery as its writer, not the reader"
);
assert_eq!(dcat["substrate"], "nursery-under-test");
assert_eq!(dcat["unit"], "collector-nursery.service");
assert_eq!(
dcat["freshness"], "live",
"the nursery is still running, so its row must read live: {:?}",
dcat["freshness_why"]
);
// ── the FACET crossed the process boundary, not just the phase ──────────────
assert!(
dcat["phase"].is_object(),
"the phase must be a structured state, not a string: {}",
dcat["phase"]
);
assert!(
!dcat["phase_label"].as_str().unwrap_or("").is_empty(),
"the phase must render: {dcat}"
);
let facet = &dcat["facet"];
assert!(
facet.is_object(),
"the FACET must cross the process boundary — a phase-only register is the hole, not \
the fix. Got: {facet}"
);
assert!(
facet.as_object().map(|o| !o.is_empty()).unwrap_or(false),
"the facet must carry the collector's own provenance, not an empty object: {facet}"
);
// ── a disabled collector is visible AS disabled ─────────────────────────────
let pxweb = rows.iter().find(|r| r["collector"] == "pxweb-jsonstat2").expect(
"a declared-off collector must still have a row — `off` is a state an \
operator asks about",
);
assert_eq!(pxweb["enabled"], false);
// ── the reader worked while the writer held redb's EXCLUSIVE lock ───────────
// Proved by construction: the nursery is still alive here, and a second WRITER over the
// same directory must fail for the lock to be genuinely held.
assert!(
korp_collectors::register::Register::open(
&data,
korp_collectors::register::Writer::this_process("test", None)
)
.is_err(),
"the nursery did NOT hold the exclusive lock, so the lock-free read proved nothing"
);
let (code, human) = run_status(&data, false);
assert_eq!(
code, 0,
"every row is live, so status must exit 0. Output:\n{human}"
);
assert!(
human.contains("dcat-ap-se"),
"the human rendering must name the collector:\n{human}"
);
assert!(
human.contains("no lock held"),
"the human rendering must say how it read:\n{human}"
);
emit_for(
"korp-collectors/register",
"separate_process_reads_live_phase_and_facet",
true,
&format!(
"serve pid {serve_pid}, status pid {reader_pid}, {} collectors, facet {}",
rows.len(),
facet
),
);
// ── ★ RED PROBE: kill the writer; the same row must stop reading as live ────
//
// Without this, a register that never checked its writer would pass every assertion
// above and then report a dead service's last phase forever.
kill_and_reap(serve);
// The row is unchanged on disk — only the writer is gone.
let (dead_code, dead_out) = run_status(&data, true);
let dv: serde_json::Value = serde_json::from_str(&dead_out).expect("status json after kill");
let dead_dcat = dv["collectors"]
.as_array()
.unwrap()
.iter()
.find(|r| r["collector"] == "dcat-ap-se")
.expect("the row must SURVIVE the writer — an operator asking what happened needs it");
assert_eq!(
dead_dcat["freshness"], "orphaned",
"a row whose writer was killed must not read as live: {dead_dcat}"
);
assert!(
dead_dcat["freshness_why"]
.as_str()
.unwrap_or("")
.contains("no longer exists"),
"the reason must name why: {}",
dead_dcat["freshness_why"]
);
assert_eq!(
dead_code, 3,
"status must EXIT NON-ZERO over a fossil row — exiting 0 there is a green that can \
never be red. Output:\n{dead_out}"
);
emit_for(
"korp-collectors/register",
"a_killed_nursery_leaves_orphaned_rows_and_status_exits_3",
true,
"writer killed → freshness=orphaned, exit 3",
);
let _ = std::fs::remove_dir_all(&data);
}
/// ★ **RED — `status` over a directory no service has used exits 4, not 0.**
///
/// The failure this guards is the one that reads identically to success: an empty answer
/// from the wrong directory. A status verb that exits 0 there tells an operator their
/// service is fine when they are looking in the wrong place.
#[test]
fn status_over_an_untouched_directory_says_so_and_exits_nonzero() {
let data = tmp();
let (code, out) = run_status(&data, false);
assert_eq!(
code, 4,
"an absent register must exit 4, got {code}. Output:\n{out}"
);
// …and with no --data-dir at all it refuses rather than guessing one.
let bare = Command::new(BIN)
.arg("status")
.env_remove("DATA_DIR")
.output()
.expect("run status with no data dir");
assert_eq!(
bare.status.code(),
Some(2),
"an undeclared data path must refuse"
);
let err = String::from_utf8_lossy(&bare.stderr);
assert!(err.contains("no data path declared"), "{err}");
emit_for(
"korp-collectors/register",
"status_exit_codes_are_a_verdict",
true,
"absent register → 4, undeclared data path → 2",
);
let _ = std::fs::remove_dir_all(&data);
}
/// ★ **RED — two nurseries over one data directory: the second REFUSES.**
///
/// Asserted through real processes because that is where it matters: two supervisors
/// interleaving rows in one register is exactly the state nobody wants resolved silently,
/// and the flock that prevents it is a property of the open, not of the code path.
#[test]
fn a_second_service_over_the_same_data_directory_is_refused_by_name() {
let data = tmp();
let first = spawn_serve(&data);
wait_for_row(&data, "dcat-ap-se", Duration::from_secs(30));
let second = Command::new(BIN)
.arg("serve")
.arg("--data-dir")
.arg(&data)
.env("COLLECTOR_NURSERY_JSON", roster(&data))
.output()
.expect("run a second service");
assert_eq!(
second.status.code(),
Some(2),
"a second nursery over one register must exit non-zero"
);
let err = String::from_utf8_lossy(&second.stderr);
assert!(
err.contains("two supervisors over one register"),
"the refusal must name WHY, not surface redb's bare error:\n{err}"
);
// The first service is untouched by the attempt: its rows are still live.
let (code, _out) = run_status(&data, false);
assert_eq!(
code, 0,
"the refused second service must not have disturbed the first"
);
kill_and_reap(first);
emit_for(
"korp-collectors/register",
"one_writer_is_enforced_across_processes",
true,
"second serve exits 2 naming the reason; first keeps live rows",
);
let _ = std::fs::remove_dir_all(&data);
}
/// ★ **RED — a roster naming a collector the build does not carry fails the SERVICE**,
/// visible as a real exit code from a real process. A partial roster that starts anyway
/// reports healthy while doing less than it was told.
#[test]
fn a_service_declaring_an_unknown_collector_exits_nonzero_naming_it() {
let data = tmp();
let out = Command::new(BIN)
.arg("serve")
.arg("--data-dir")
.arg(&data)
.env(
"COLLECTOR_NURSERY_JSON",
serde_json::json!({
"data_dir": &data,
"collectors": [{ "name": "not-a-real-collector" }]
})
.to_string(),
)
.output()
.expect("run serve with a bad roster");
assert_eq!(out.status.code(), Some(1), "a bad roster must not exit 0");
let err = String::from_utf8_lossy(&out.stderr);
assert!(err.contains("not-a-real-collector"), "{err}");
assert!(err.contains("Refusing to start a partial roster"), "{err}");
emit_for(
"korp-collectors/nursery",
"unknown_collector_fails_the_service",
true,
"serve exits 1 naming the collector",
);
let _ = std::fs::remove_dir_all(&data);
}
/// The heartbeat keeps the row moving, so a reader can tell a working collector from a
/// stalled one. Asserted as a **strictly advancing publish counter observed from outside**,
/// which a single startup publish cannot produce.
#[test]
fn the_register_is_refreshed_while_the_collector_runs_so_a_reader_can_see_it_advance() {
let data = tmp();
let serve = spawn_serve(&data);
let first = wait_for_row(&data, "dcat-ap-se", Duration::from_secs(30));
let seq0 = first["collectors"]
.as_array()
.unwrap()
.iter()
.find(|r| r["collector"] == "dcat-ap-se")
.unwrap()["seq"]
.as_u64()
.unwrap();
// heartbeat_ms is 300 in the roster, so a few beats fit comfortably here.
let t0 = Instant::now();
let mut seq1 = seq0;
while t0.elapsed() < Duration::from_secs(15) && seq1 <= seq0 {
std::thread::sleep(Duration::from_millis(200));
let (_c, out) = run_status(&data, true);
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&out) {
if let Some(r) = v["collectors"]
.as_array()
.and_then(|a| a.iter().find(|r| r["collector"] == "dcat-ap-se"))
{
seq1 = r["seq"].as_u64().unwrap_or(seq0);
}
}
}
kill_and_reap(serve);
assert!(
seq1 > seq0,
"the publish counter did not advance ({seq0} → {seq1}) — a register that is written \
once at startup is a transition log, not a live view"
);
emit_for(
"korp-collectors/register",
"heartbeat_advances_the_row_for_an_outside_reader",
true,
&format!("seq {seq0} → {seq1} observed from a separate process"),
);
let _ = std::fs::remove_dir_all(&data);
}