src/register.rs
//! **The live register — where a collector's phase AND facet are readable from ANOTHER
//! PROCESS** (feature `daemon`).
//!
//! # The hole this closes
//!
//! A collector's live state used to be readable only from inside the process supervising
//! it: korp reads `live_facet` out of its in-memory supervisor, documented as *"never from
//! a file snapshot"* (`korp/src/collectors_daemon.rs:412`). While that holds, a service
//! outside the process **cannot see the state of the collectors it is supposed to nurse**
//! — which is the whole difference between a supervisor and a thread pool. A status verb,
//! an infra-matrix cell, an operator on the box and a second service all ask the same
//! question and none of them is the process that owns the answer.
//!
//! The fix is the one this crate already uses for *"is this step finished?"*: **the
//! producer knows, the producer writes it down, and the reader reads the register.**
//! [`crate::state`] does that for completion facts. This does it for live state, in the
//! same embedded store, through the same lock-free snapshot reader
//! ([`crate::state::open_snapshot_db`]) — a second mechanism would be a second set of
//! concurrency bounds to keep in agreement.
//!
//! # Two stores, because they answer two different questions
//!
//! | | [`crate::state`] | this |
//! |---|---|---|
//! | file | `<collector data>/collector-state.redb` | `<service data>/collector-register.redb` |
//! | scope | one collector's own chain | **every collector one service nurses** |
//! | written | once, when a step finishes | on every phase transition |
//! | means | this artefact is durable and provenanced | this process is nursing this collector, and here is what it is doing |
//!
//! A completion fact outlives the process; live state does not, and conflating them is
//! how a stopped collector reads as busy forever.
//!
//! # Why a row carries WHO wrote it
//!
//! Live state read out of a file has one failure mode a read out of memory cannot have:
//! **the writer can be gone.** A row saying `Cutting sweden 60%` is true only while the
//! process that wrote it still exists; after a `SIGKILL` it is a fossil, and a reader that
//! prints it is reporting work that nobody is doing. So every row carries the writer's
//! identity — pid, the process's own start time, and the boot id — and
//! [`RegisterReader`] answers with a [`Freshness`] beside every row rather than handing
//! back a phase and letting each caller decide whether to believe it.
//!
//! Start time and boot id are both needed, and neither is decoration: a pid alone is
//! reused within minutes on a busy box, and pid+start-time repeats across a reboot
//! because start time is measured from boot.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use redb::{Database, ReadableTable, TableDefinition};
use serde::{Deserialize, Serialize};
use crate::daemon::{Collector, Phase};
/// The register file, inside the SERVICE's data directory (not a collector's).
pub const REGISTER_DB_FILE: &str = "collector-register.redb";
/// collector name → the JSON [`LiveRow`].
const LIVE: TableDefinition<&str, &str> = TableDefinition::new("live");
// ─────────────────────────────────────────────────────────────────────────────
// Who wrote a row
// ─────────────────────────────────────────────────────────────────────────────
/// **The identity of the process that published a row**, precise enough that a reader can
/// tell "still running" from "died holding the pen".
///
/// All three fields are required together. `pid` alone is reused; `pid` + `started_ticks`
/// is measured from boot and therefore repeats across a reboot; `boot_id` closes that.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Writer {
/// The publishing process's pid.
pub pid: u32,
/// Its start time in clock ticks since boot (`/proc/<pid>/stat` field 22). `None`
/// where `/proc` does not exist — and then liveness is [`Freshness::Undecidable`],
/// never assumed.
pub started_ticks: Option<u64>,
/// The kernel's boot id, so a pid+start-time pair cannot be matched against a
/// different boot.
pub boot_id: Option<String>,
/// How the process was running the collector — `"nursery"`, `"systemd"`, `"korp"`.
/// Free text, for an operator, never matched on.
pub substrate: String,
/// The unit or job name, when the substrate has one (`collector-dcat.service`).
pub unit: Option<String>,
}
impl Writer {
/// This process, as it should be recorded.
pub fn this_process(substrate: impl Into<String>, unit: Option<String>) -> Self {
Writer {
pid: std::process::id(),
started_ticks: proc_started_ticks(std::process::id()),
boot_id: boot_id(),
substrate: substrate.into(),
unit,
}
}
/// Is the process that wrote this row still the same live process?
///
/// Three answers, never two. `Undecidable` is not "alive": on a host without `/proc`
/// the question cannot be answered, and answering it anyway is how a fossil row gets
/// printed as live work.
pub fn liveness(&self) -> Freshness {
let (Some(want), Some(want_boot)) = (self.started_ticks, self.boot_id.as_deref()) else {
return Freshness::Undecidable {
why: "the row carries no process start time or boot id, so whether its writer \
still exists cannot be decided from it"
.to_string(),
};
};
match boot_id() {
None => Freshness::Undecidable {
why: "this host publishes no boot id, so a pid+start-time pair cannot be \
pinned to the boot that produced it"
.to_string(),
},
Some(now_boot) if now_boot != want_boot => Freshness::Orphaned {
why: format!(
"the row was written before a reboot (boot {want_boot}, now {now_boot}) — \
pid {} cannot be the process that wrote it",
self.pid
),
},
Some(_) => match proc_started_ticks(self.pid) {
None => Freshness::Orphaned {
why: format!(
"pid {} no longer exists — the process that published this row is gone, \
so the phase it left behind is a fossil, not work in progress",
self.pid
),
},
Some(now) if now != want => Freshness::Orphaned {
why: format!(
"pid {} exists but started at tick {now}, not {want} — the pid was \
REUSED by a different process, and this row's writer is gone",
self.pid
),
},
Some(_) => Freshness::Live,
},
}
}
}
/// **Whether a row's writer is still there** — the judgement a reader of live state must
/// make and a reader of in-memory state never had to.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Freshness {
/// The publishing process is alive. The phase is current.
Live,
/// The publishing process is gone. The row is the last thing it managed to say, and
/// is reported AS THAT — never as the current phase.
Orphaned { why: String },
/// Liveness could not be determined on this host. Not green, not a fault: stated.
Undecidable { why: String },
}
impl Freshness {
/// True only for [`Freshness::Live`]. `Undecidable` is deliberately not live.
pub fn is_live(&self) -> bool {
matches!(self, Freshness::Live)
}
/// The stable token a test or a matrix cell asserts on.
pub fn kind(&self) -> &'static str {
match self {
Freshness::Live => "live",
Freshness::Orphaned { .. } => "orphaned",
Freshness::Undecidable { .. } => "undecidable",
}
}
pub fn why(&self) -> Option<&str> {
match self {
Freshness::Live => None,
Freshness::Orphaned { why } | Freshness::Undecidable { why } => Some(why),
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The row
// ─────────────────────────────────────────────────────────────────────────────
/// **One collector's live state, as published.**
///
/// The facet is here, beside the phase, and that is the point: a phase alone says
/// `cutting 60%`, and the facet is what says *1 599 775 tiles into `nordisk-maps`*. A
/// register that carried only the phase would leave every out-of-process substrate
/// showing progress with no provenance, which is the state this module exists to end.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct LiveRow {
/// The collector name — the row key, repeated in the value so a row read alone is
/// self-describing.
pub collector: String,
/// Whether the operator has it switched on. A disabled collector still gets a row:
/// "off" is a state an operator asks about.
pub enabled: bool,
/// The state machine's position.
pub phase: Phase,
/// The collector's own health verdict, which is not always derivable from the phase.
pub healthy: bool,
/// **The collector's own provenance blob** — whatever [`Collector::facet`] returns.
pub facet: serde_json::Value,
/// Who published this.
pub writer: Writer,
/// When (unix seconds).
pub updated_unix: u64,
/// Monotonic per-collector publish counter, so a reader can tell a re-published
/// identical phase from a stalled one.
pub seq: u64,
}
impl LiveRow {
/// A one-line rendering for a status verb.
pub fn line(&self, fresh: &Freshness) -> String {
let facet = serde_json::to_string(&self.facet).unwrap_or_else(|_| "?".into());
format!(
"{:<16} {:<8} {:<22} {} pid {} · {}{}",
self.collector,
if self.enabled { "enabled" } else { "off" },
self.phase.label(),
if self.healthy { "healthy" } else { "unhealthy" },
self.writer.pid,
facet,
match fresh {
Freshness::Live => String::new(),
f => format!(" [{}: {}]", f.kind(), f.why().unwrap_or("")),
}
)
}
}
/// A row plus the judgement on its writer — what every reader actually wants, so no
/// caller can read the phase and forget to ask whether anyone is still behind it.
#[derive(Clone, Debug)]
pub struct LiveView {
pub row: LiveRow,
pub freshness: Freshness,
}
impl LiveView {
/// The phase a reader may act on: `Some` only while the writer is alive.
///
/// This is the shape rather than a bare getter because the alternative —
/// `view.row.phase` — reads the same in code that checked the writer and code that
/// did not.
pub fn current_phase(&self) -> Option<&Phase> {
if self.freshness.is_live() {
Some(&self.row.phase)
} else {
None
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The writer side
// ─────────────────────────────────────────────────────────────────────────────
/// **The single writer of the live register**, held by whichever process is nursing the
/// collectors.
///
/// redb takes `flock(LOCK_EX | LOCK_NB)` on open, so a second nursery over the same data
/// directory fails here rather than interleaving rows — which is correct, and the message
/// says so.
#[derive(Debug)]
pub struct Register {
db: Database,
path: PathBuf,
writer: Writer,
seq: std::sync::atomic::AtomicU64,
}
impl Register {
/// Open (creating if absent) the register in the service's `data_dir`.
pub fn open(data_dir: &Path, writer: Writer) -> Result<Self> {
std::fs::create_dir_all(data_dir)
.with_context(|| format!("mkdir {}", data_dir.display()))?;
let path = data_dir.join(REGISTER_DB_FILE);
let db = Database::create(&path).with_context(|| {
format!(
"open the collector register at {} for WRITING. redb holds an exclusive \
flock, so this fails while another nursery has the same data directory — \
two supervisors over one register is exactly the situation nobody wants \
resolved silently. Readers use RegisterReader, which takes no lock",
path.display()
)
})?;
let w = db.begin_write()?;
{
let _ = w.open_table(LIVE)?;
}
w.commit()?;
Ok(Self {
db,
path,
writer,
seq: std::sync::atomic::AtomicU64::new(0),
})
}
pub fn path(&self) -> &Path {
&self.path
}
/// **Publish a collector's live state, SAMPLED FROM THE COLLECTOR ITSELF.**
///
/// Taking `&dyn Collector` rather than the three values is what makes the phase and
/// the facet impossible to disagree: there is no call site that can publish a new
/// phase while leaving last pass's facet in place, because there is no argument for
/// it to pass. That is the whole reason the register is written here and not by each
/// substrate.
pub fn publish(&self, c: &dyn Collector) -> Result<LiveRow> {
let row = LiveRow {
collector: c.name().to_string(),
enabled: c.enabled(),
phase: c.phase(),
healthy: c.healthy(),
facet: c.facet(),
writer: self.writer.clone(),
updated_unix: now_unix(),
seq: self.seq.fetch_add(1, std::sync::atomic::Ordering::SeqCst),
};
self.write_row(&row)?;
Ok(row)
}
/// Write a row verbatim. `pub` so a substrate that holds a collector's state but not
/// the collector (a re-publish after a restart, a test's red probe) can still write
/// one — and named so that a diff shows which of the two happened.
pub fn write_row(&self, row: &LiveRow) -> Result<()> {
let json = serde_json::to_string(row).context("serialize the live row")?;
let w = self.db.begin_write()?;
{
let mut t = w.open_table(LIVE)?;
t.insert(row.collector.as_str(), json.as_str())?;
}
w.commit().context("commit the live row")?;
Ok(())
}
/// **Record that a collector's run ENDED, and why.**
///
/// The row stays — an operator asking "what happened to dcat" must get an answer, and
/// a deleted row cannot give one. What changes is the phase, so a finished failure is
/// not left looking like work in progress.
pub fn retire(&self, c: &dyn Collector, phase: Phase) -> Result<()> {
let mut row = LiveRow {
collector: c.name().to_string(),
enabled: c.enabled(),
phase: c.phase(),
healthy: c.healthy(),
facet: c.facet(),
writer: self.writer.clone(),
updated_unix: now_unix(),
seq: self.seq.fetch_add(1, std::sync::atomic::Ordering::SeqCst),
};
row.phase = phase;
row.healthy = false;
self.write_row(&row)
}
/// Read back what this process published — for a self-check, never as the path a
/// separate process uses (that is [`RegisterReader`]).
pub fn get(&self, collector: &str) -> Result<Option<LiveRow>> {
let r = self.db.begin_read()?;
let t = r.open_table(LIVE)?;
Ok(t.get(collector)?
.and_then(|v| serde_json::from_str::<LiveRow>(v.value()).ok()))
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The reader side — a DIFFERENT PROCESS, no lock
// ─────────────────────────────────────────────────────────────────────────────
/// **The out-of-process reader.** Opens the register `O_RDONLY`, takes no lock, and works
/// while the nursery is writing.
///
/// This type is the answer to H6. A `korp-collectors status`, an infra-matrix cell, or a
/// foreign project's own dashboard opens one of these and sees the phase *and the facet*
/// of every collector some other process is nursing.
#[derive(Debug)]
pub struct RegisterReader {
db: Database,
path: PathBuf,
}
impl RegisterReader {
/// Open the register in `data_dir`. `Ok(None)` when no service has ever run there —
/// which is an answer, and must not read as a failure.
pub fn open(data_dir: &Path) -> Result<Option<Self>> {
let path = data_dir.join(REGISTER_DB_FILE);
Ok(crate::state::open_snapshot_db(&path)?.map(|db| Self { db, path }))
}
pub fn path(&self) -> &Path {
&self.path
}
/// Every collector in the register, name-ordered, each with its writer judged.
pub fn all(&self) -> Result<Vec<LiveView>> {
let r = self.db.begin_read()?;
let t = match r.open_table(LIVE) {
Ok(t) => t,
// A register that exists but holds no table yet holds no collectors.
Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
Err(e) => return Err(e.into()),
};
let mut out = Vec::new();
for row in t.iter()? {
let (_, v) = row?;
if let Ok(r) = serde_json::from_str::<LiveRow>(v.value()) {
let freshness = r.writer.liveness();
out.push(LiveView { row: r, freshness });
}
}
out.sort_by(|a, b| a.row.collector.cmp(&b.row.collector));
Ok(out)
}
/// One collector, with its writer judged.
pub fn get(&self, collector: &str) -> Result<Option<LiveView>> {
Ok(self
.all()?
.into_iter()
.find(|v| v.row.collector == collector))
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Process identity, read from /proc
// ─────────────────────────────────────────────────────────────────────────────
/// A process's start time in clock ticks since boot — `/proc/<pid>/stat` field 22.
///
/// `None` means the pid does not exist OR `/proc` is unavailable, and the two are told
/// apart by the caller having asked about *itself* first: a process that cannot read its
/// own start time is on a host where liveness is [`Freshness::Undecidable`].
///
/// The parse starts after the LAST `)`, because field 2 is the executable name and it may
/// contain both spaces and parentheses — splitting the whole line on whitespace is the
/// classic way to read the wrong field.
fn proc_started_ticks(pid: u32) -> Option<u64> {
let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
let after = &stat[stat.rfind(')')? + 1..];
// After `)` the fields are 3..; field 22 is therefore index 19 here.
after.split_whitespace().nth(19)?.parse().ok()
}
/// The kernel's boot id, so a pid+start-time pair belongs to a known boot.
fn boot_id() -> Option<String> {
std::fs::read_to_string("/proc/sys/kernel/random/boot_id")
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
}
fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
// ─────────────────────────────────────────────────────────────────────────────
// RED PROBES
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use crate::daemon::{CollectorConfig, CollectorState, ConfigSchema};
use serde_json::json;
use std::sync::Mutex;
/// A collector whose phase and facet a test can move, so publishing can be observed
/// sampling them rather than echoing what a caller passed.
struct Probe {
phase: Mutex<Phase>,
tiles: Mutex<u64>,
}
impl Probe {
fn new() -> Self {
Self {
phase: Mutex::new(Phase::Offline),
tiles: Mutex::new(0),
}
}
}
impl CollectorConfig for Probe {
fn name(&self) -> &str {
"probe"
}
fn enabled(&self) -> bool {
true
}
fn schema(&self) -> ConfigSchema {
ConfigSchema::new(vec![])
}
}
impl CollectorState for Probe {
fn phase(&self) -> Phase {
self.phase.lock().unwrap().clone()
}
fn facet(&self) -> serde_json::Value {
json!({ "tiles_written": *self.tiles.lock().unwrap() })
}
}
#[async_trait::async_trait]
impl Collector for Probe {
async fn run(self: std::sync::Arc<Self>, _ctx: crate::daemon::CollectorCtx) {}
}
fn reg(dir: &Path) -> Register {
Register::open(dir, Writer::this_process("test", None)).unwrap()
}
/// The green the reds below are measured against: a published row is readable through
/// a lock-free reader, **carries the facet**, and is judged live.
#[test]
fn a_published_row_is_readable_with_its_facet_while_the_writer_holds_the_store() {
let d = tempfile::tempdir().unwrap();
let p = Probe::new();
let r = reg(d.path());
*p.phase.lock().unwrap() = Phase::Cutting {
region: "sweden".into(),
pct: 0.6,
};
*p.tiles.lock().unwrap() = 1_599_775;
r.publish(&p).unwrap();
// A SECOND writer must still be refused — one writer is the invariant.
assert!(
Register::open(d.path(), Writer::this_process("test", None)).is_err(),
"two nurseries were allowed over one register"
);
let reader = RegisterReader::open(d.path())
.unwrap()
.expect("register exists");
let v = reader.get("probe").unwrap().expect("the row is readable");
assert_eq!(v.freshness.kind(), "live", "{:?}", v.freshness.why());
assert_eq!(
v.current_phase(),
Some(&Phase::Cutting {
region: "sweden".into(),
pct: 0.6
})
);
assert_eq!(
v.row.facet["tiles_written"], 1_599_775,
"the FACET must survive the round trip — a phase-only register is the hole, \
not the fix"
);
assert_eq!(v.row.writer.pid, std::process::id());
}
/// ★ **RED 1 — a row whose writer is DEAD must not read as the current phase.**
///
/// The row is well-formed and says `Online`. Only the writer's absence makes it a
/// fossil, which is the failure mode a file-backed register has and an in-memory one
/// does not.
#[test]
fn a_row_from_a_dead_writer_is_orphaned_and_yields_no_current_phase() {
let d = tempfile::tempdir().unwrap();
let r = reg(d.path());
// pid 0 never exists as a process on Linux, so this is a writer that is gone.
let mut row = LiveRow {
collector: "ghost".into(),
enabled: true,
phase: Phase::Online,
healthy: true,
facet: json!({ "rows": 42 }),
writer: Writer::this_process("test", None),
updated_unix: now_unix(),
seq: 0,
};
row.writer.pid = 0;
r.write_row(&row).unwrap();
let reader = RegisterReader::open(d.path()).unwrap().unwrap();
let v = reader.get("ghost").unwrap().unwrap();
assert_eq!(v.freshness.kind(), "orphaned", "{:?}", v.freshness);
assert!(
v.current_phase().is_none(),
"an orphaned row must not hand back a phase"
);
assert!(
v.freshness.why().unwrap().contains("no longer exists"),
"the reason must name WHY: {:?}",
v.freshness.why()
);
// The row itself is still there — an operator asking what happened gets an answer.
assert_eq!(v.row.phase, Phase::Online);
}
/// ★ **RED 2 — a REUSED pid is not the writer.** The pid exists (this process), but
/// it started at a different tick, and that must be caught: on a busy box a pid is
/// recycled in minutes, and matching on pid alone turns a stranger's process into
/// evidence that our collector is running.
#[test]
fn a_reused_pid_is_detected_by_its_start_time() {
let d = tempfile::tempdir().unwrap();
let r = reg(d.path());
let mut row = LiveRow {
collector: "reused".into(),
enabled: true,
phase: Phase::Indexing,
healthy: false,
facet: json!({}),
writer: Writer::this_process("test", None),
updated_unix: now_unix(),
seq: 0,
};
// Same pid (alive), different start tick.
row.writer.started_ticks = Some(row.writer.started_ticks.unwrap() + 1);
r.write_row(&row).unwrap();
let v = RegisterReader::open(d.path())
.unwrap()
.unwrap()
.get("reused")
.unwrap()
.unwrap();
assert_eq!(v.freshness.kind(), "orphaned", "{:?}", v.freshness);
assert!(
v.freshness.why().unwrap().contains("REUSED"),
"{:?}",
v.freshness.why()
);
}
/// ★ **RED 3 — a row written before a reboot is not live**, even though pid and start
/// tick both match. Start time is measured FROM boot, so the pair repeats.
#[test]
fn a_row_from_before_a_reboot_is_orphaned_even_when_pid_and_start_tick_match() {
let d = tempfile::tempdir().unwrap();
let r = reg(d.path());
let mut row = LiveRow {
collector: "prev-boot".into(),
enabled: true,
phase: Phase::Online,
healthy: true,
facet: json!({}),
writer: Writer::this_process("test", None),
updated_unix: now_unix(),
seq: 0,
};
row.writer.boot_id = Some("00000000-0000-0000-0000-000000000000".into());
r.write_row(&row).unwrap();
let v = RegisterReader::open(d.path())
.unwrap()
.unwrap()
.get("prev-boot")
.unwrap()
.unwrap();
assert_eq!(v.freshness.kind(), "orphaned", "{:?}", v.freshness);
assert!(
v.freshness.why().unwrap().contains("reboot"),
"{:?}",
v.freshness.why()
);
}
/// ★ **RED 4 — liveness must be UNDECIDABLE, not live, when it cannot be checked.**
/// A row missing the process identity is the shape a foreign or older writer produces.
#[test]
fn a_row_without_process_identity_is_undecidable_and_not_live() {
let d = tempfile::tempdir().unwrap();
let r = reg(d.path());
let mut row = LiveRow {
collector: "anon".into(),
enabled: true,
phase: Phase::Online,
healthy: true,
facet: json!({}),
writer: Writer::this_process("test", None),
updated_unix: now_unix(),
seq: 0,
};
row.writer.started_ticks = None;
r.write_row(&row).unwrap();
let v = RegisterReader::open(d.path())
.unwrap()
.unwrap()
.get("anon")
.unwrap()
.unwrap();
assert_eq!(v.freshness.kind(), "undecidable", "{:?}", v.freshness);
assert!(
!v.freshness.is_live(),
"undecidable must never count as live"
);
assert!(v.current_phase().is_none());
}
/// **Publishing SAMPLES the collector**, so phase and facet cannot drift apart. The
/// probe's facet moves between publishes and the register follows it — with no
/// argument at the call site that could have carried a stale one.
#[test]
fn publish_samples_the_collector_so_phase_and_facet_move_together() {
let d = tempfile::tempdir().unwrap();
let p = Probe::new();
let r = reg(d.path());
r.publish(&p).unwrap();
let first = RegisterReader::open(d.path())
.unwrap()
.unwrap()
.get("probe")
.unwrap()
.unwrap();
assert_eq!(first.row.phase, Phase::Offline);
assert_eq!(first.row.facet["tiles_written"], 0);
*p.phase.lock().unwrap() = Phase::Online;
*p.tiles.lock().unwrap() = 7;
r.publish(&p).unwrap();
let second = RegisterReader::open(d.path())
.unwrap()
.unwrap()
.get("probe")
.unwrap()
.unwrap();
assert_eq!(second.row.phase, Phase::Online);
assert_eq!(second.row.facet["tiles_written"], 7);
assert!(
second.row.seq > first.row.seq,
"the publish counter must advance"
);
}
/// A retired collector stops looking like work in progress, and keeps its answer.
#[test]
fn retiring_a_collector_replaces_the_phase_and_keeps_the_row() {
let d = tempfile::tempdir().unwrap();
let p = Probe::new();
let r = reg(d.path());
*p.phase.lock().unwrap() = Phase::Cutting {
region: "sweden".into(),
pct: 0.4,
};
*p.tiles.lock().unwrap() = 99;
r.publish(&p).unwrap();
r.retire(
&p,
Phase::Failed {
reason: "planet unreadable".into(),
},
)
.unwrap();
let v = RegisterReader::open(d.path())
.unwrap()
.unwrap()
.get("probe")
.unwrap()
.unwrap();
assert_eq!(
v.row.phase,
Phase::Failed {
reason: "planet unreadable".into()
},
"a dead run must not be left mid-cut"
);
assert!(!v.row.healthy);
assert_eq!(
v.row.facet["tiles_written"], 99,
"the last facet is kept — it is the answer"
);
}
/// ★ The reader must not write a byte back to the register file.
#[test]
fn the_reader_never_writes_back_to_the_register() {
let d = tempfile::tempdir().unwrap();
let p = Probe::new();
let w = reg(d.path());
w.publish(&p).unwrap();
let f = d.path().join(REGISTER_DB_FILE);
let before = std::fs::read(&f).unwrap();
{
let r = RegisterReader::open(d.path()).unwrap().unwrap();
assert_eq!(r.all().unwrap().len(), 1);
}
assert_eq!(
before,
std::fs::read(&f).unwrap(),
"the reader modified the register on disk — it may only scribble on its RAM snapshot"
);
// …and the writer still works afterwards.
*p.phase.lock().unwrap() = Phase::Online;
w.publish(&p).unwrap();
assert_eq!(w.get("probe").unwrap().unwrap().phase, Phase::Online);
}
/// An empty data directory is an answer, not an error.
#[test]
fn a_register_that_has_never_been_written_reads_as_absent() {
let d = tempfile::tempdir().unwrap();
assert!(RegisterReader::open(d.path()).unwrap().is_none());
}
/// The `/proc/<pid>/stat` field walk must survive an executable name containing
/// spaces and parentheses — the classic way to read the wrong field.
#[test]
fn this_process_can_read_its_own_start_time() {
let mine = proc_started_ticks(std::process::id());
assert!(
mine.is_some(),
"cannot read own /proc start time — liveness would be undecidable"
);
assert!(
proc_started_ticks(0).is_none(),
"pid 0 must not resolve to a start time"
);
}
}