src/nursery.rs
//! **The nursery — one service, many collectors** (feature `service`).
//!
//! # What this is for
//!
//! > *"Vill kunna återanvända collector som en tjänst som kör många collectorer. Sätter
//! > man upp en helt ny tjänst kan man installera collector som system, lägga in sina egna
//! > collectorer utan att hela OpenStreetMap dras med."*
//!
//! A foreign project installs this, declares which collectors it wants, and gets a
//! supervised service. It brings no OSM: the geometry stack is behind `osm-daemon`, and a
//! service built without that feature has a registry of pure-parser collectors and no
//! planet anywhere in its graph.
//!
//! # What the nursery owns, and what it deliberately does not
//!
//! | it owns | it does NOT own |
//! |---|---|
//! | one supervised run per declared collector, restarted with backoff | the unit file, the account, the data directory — [`crate::runtime_systemd`] routes those through Skidbladnir |
//! | publishing every collector's phase AND facet to the [register](crate::register) | *where the collected rows land* — see below |
//! | one process-wide network wait, and one verdict when the work dies | either decision itself: both come from `skidbladnir::readiness` |
//!
//! **The sink is deliberately absent.** Where a collector's rows are written, and in what
//! format, is a decision that has not been made. A nursery that quietly picked a warehouse
//! would be making it, so [`CollectorCtx::in_memory`](crate::daemon::CollectorCtx::in_memory)'s
//! sinks stay in place and the log says so at startup. Everything else — the fetch, the
//! phase, the facet, the restart, the register — is real.
//!
//! # Why the data path is declared and never derived
//!
//! [`NurseryConfig::data_dir`] is required. There is no `$HOME` default, and
//! [`Nursery::new`] refuses a data directory that is not under the home of the account
//! the process is actually running as.
//!
//! That guard is there because of a measured failure: an install run in system scope by a
//! login user put 1 605 675 tiles in `/home/<login-user>/.korp` while the uninstall swept
//! `/home/korp/.korp` and reported success (`korp/src/service.rs:205-206`). The truth
//! about where a service's data lives is its unit's `User=` and that account's passwd
//! entry — never `$HOME`, which is whatever the invoking shell had. Skidbladnir's layout
//! resolves it from passwd; this refuses anything that disagrees with it.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use skidbladnir::readiness::{self, Needs, NetworkReady, WorkStatus};
use crate::daemon::{
Collector, CollectorCtx, Fetcher, LogSink, Phase, Shutdown, daemon_with_config,
};
use crate::register::{Register, Writer};
/// The env var carrying the nursery's own declaration, for a substrate that has no place
/// to put a file. Mirrors [`crate::runtime_systemd::CONFIG_ENV`], which carries ONE
/// collector's config; this carries the whole roster.
pub const NURSERY_CONFIG_ENV: &str = "COLLECTOR_NURSERY_JSON";
/// How long the nursery waits for the network before reporting it absent. Skidbladnir's
/// own default; named here so the value has one owner.
pub const NETWORK_WAIT: Duration = readiness::DEFAULT_NETWORK_WAIT;
// ─────────────────────────────────────────────────────────────────────────────
// The declaration
// ─────────────────────────────────────────────────────────────────────────────
/// **One collector the operator wants nursed.**
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct CollectorDecl {
/// The registry name (`korp-collectors list` prints them).
pub name: String,
/// Whether to run it. A disabled collector still gets a register row — "off" is a
/// state an operator asks about, and a missing row cannot express it.
#[serde(default = "yes")]
pub enabled: bool,
/// The collector's own config, exactly as [`crate::daemon::CollectorConfig::schema`]
/// describes it.
#[serde(default)]
pub config: serde_json::Value,
/// How many times a failed run is restarted before the nursery gives up on this
/// collector. `None` = forever, which is the right default for a daemon whose source
/// is a remote service that comes and goes.
#[serde(default)]
pub max_restarts: Option<u32>,
}
fn yes() -> bool {
true
}
impl CollectorDecl {
pub fn new(name: impl Into<String>) -> Self {
Self {
name: name.into(),
enabled: true,
config: serde_json::json!({}),
max_restarts: None,
}
}
pub fn with_config(mut self, config: serde_json::Value) -> Self {
self.config = config;
self
}
pub fn disabled(mut self) -> Self {
self.enabled = false;
self
}
pub fn giving_up_after(mut self, n: u32) -> Self {
self.max_restarts = Some(n);
self
}
}
/// **Everything the nursery is told.** Nothing here is guessed and nothing is read from
/// the ambient environment except by an explicit call to [`NurseryConfig::from_env`].
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct NurseryConfig {
/// The service's data root. Required — see the module note on why there is no default.
pub data_dir: PathBuf,
/// The roster.
pub collectors: Vec<CollectorDecl>,
/// What the service needs from the host before its work can run. Honoured by
/// `skidbladnir::readiness::await_network`, not by a wait written here.
#[serde(default)]
pub needs: NeedsDecl,
/// First restart delay; doubles per consecutive failure up to [`Self::backoff_max`].
#[serde(default = "default_backoff_min_ms")]
pub backoff_min_ms: u64,
#[serde(default = "default_backoff_max_ms")]
pub backoff_max_ms: u64,
/// How often a still-running collector's row is refreshed, so a reader can tell a
/// working collector from a stalled one by the row's own age.
#[serde(default = "default_heartbeat_ms")]
pub heartbeat_ms: u64,
/// Substrate label recorded on every register row, for an operator reading it.
#[serde(default = "default_substrate")]
pub substrate: String,
/// The unit name, when a substrate has one.
#[serde(default)]
pub unit: Option<String>,
}
fn default_backoff_min_ms() -> u64 {
1_000
}
fn default_backoff_max_ms() -> u64 {
300_000
}
fn default_heartbeat_ms() -> u64 {
5_000
}
fn default_substrate() -> String {
"nursery".into()
}
/// Serde mirror of [`Needs`], which is Skidbladnir's type and carries no serde derive.
/// A mirror rather than a fork: it converts, it does not decide.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NeedsDecl {
/// Local work only — start immediately.
#[default]
Nothing,
/// A usable network. The wait itself is Skidbladnir's.
Network,
}
impl From<NeedsDecl> for Needs {
fn from(d: NeedsDecl) -> Needs {
match d {
NeedsDecl::Nothing => Needs::Nothing,
NeedsDecl::Network => Needs::Network,
}
}
}
impl NurseryConfig {
/// A config for `data_dir` with `collectors` and no other opinion.
pub fn new(data_dir: impl Into<PathBuf>, collectors: Vec<CollectorDecl>) -> Self {
Self {
data_dir: data_dir.into(),
collectors,
needs: NeedsDecl::Nothing,
backoff_min_ms: default_backoff_min_ms(),
backoff_max_ms: default_backoff_max_ms(),
heartbeat_ms: default_heartbeat_ms(),
substrate: default_substrate(),
unit: None,
}
}
pub fn needing_network(mut self) -> Self {
self.needs = NeedsDecl::Network;
self
}
/// Read the declaration a substrate handed over: `--config <file>` (a ConfigMap
/// mount) or [`NURSERY_CONFIG_ENV`] (a systemd `Environment=`). `data_dir` may be
/// overridden by `$DATA_DIR`, which is what Skidbladnir's layout declares and what its
/// uninstall erases.
///
/// Absent everything, this REFUSES. A nursery that invents a roster runs nothing an
/// operator asked for, and a nursery that invents a data directory puts bytes
/// somewhere the uninstall will never look.
pub fn from_env(config_file: Option<&str>) -> Result<Self> {
let raw = match config_file {
Some(f) => std::fs::read_to_string(f)
.with_context(|| format!("read the nursery declaration from {f}"))?,
None => std::env::var(NURSERY_CONFIG_ENV).map_err(|_| {
anyhow::anyhow!(
"no nursery declaration: pass --config <file> or set ${NURSERY_CONFIG_ENV}. \
Refusing to invent a roster — a service that picks its own collectors is \
running something nobody asked for"
)
})?,
};
let mut cfg: NurseryConfig =
serde_json::from_str(&raw).context("parse the nursery declaration as JSON")?;
if let Ok(d) = std::env::var("DATA_DIR") {
if !d.trim().is_empty() {
cfg.data_dir = PathBuf::from(d);
}
}
Ok(cfg)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The data-path guard
// ─────────────────────────────────────────────────────────────────────────────
/// **Check that a data path agrees with the account the process is actually running as.**
///
/// Pure, and separate from [`Nursery::new`], because the whole reason this exists is a
/// disagreement between two answers to "where does the data live" — so the check has to be
/// callable with both answers written down, in a test, on a box with no service accounts.
///
/// `passwd_home` is the home of the running uid as `/etc/passwd` gives it — the SAME
/// source Skidbladnir's layout and chown guard read, so this cannot disagree with the
/// installer.
pub fn data_dir_belongs_to_account(
data_dir: &Path,
uid: u32,
user: &str,
passwd_home: Option<&Path>,
) -> Result<()> {
// Root installs into system paths on purpose; the failure this guards against is a
// LOGIN user's home being used by a service that declares another account.
if uid == 0 {
return Ok(());
}
let Some(home) = passwd_home else {
bail!(
"uid {uid} ({user}) has no home in passwd, so `{}` cannot be checked against the \
account that will own it. The truth about where a service's data lives is its \
unit's User= and that account's passwd entry",
data_dir.display()
);
};
if data_dir.starts_with(home) {
return Ok(());
}
// Another *service* home is the shape of the measured bug: the unit declares User=korp,
// the install ran as a login user, and the two homes are siblings.
bail!(
"the declared data path `{}` is not under the home of the account this process runs as \
(uid {uid} `{user}`, home `{}`).\n\
This is refused because it already happened: an install run in system scope by a login \
user wrote 1 605 675 tiles into /home/<login-user>/.korp while the uninstall swept \
/home/korp/.korp and reported success. The data path and the unit's User= must name the \
same account — install through Skidbladnir, which creates both, rather than passing a \
path from the invoking shell",
data_dir.display(),
home.display()
)
}
/// The running uid's `(uid, name, home)` out of `/etc/passwd`.
///
/// Every field comes from Skidbladnir — `current_euid` and the ONE passwd parser that the
/// chown guard, the drop-in scanner and [`Layout::for_service_user`](skidbladnir::layout::Layout::for_service_user)
/// already read. A second parser here would be a second answer to the question the guard
/// exists to settle.
fn running_account() -> (u32, String, Option<PathBuf>) {
let uid = skidbladnir::systemd::current_euid();
match skidbladnir::systemd::passwd_homes_of(Path::new("/etc/passwd"))
.into_iter()
.find(|(_, u, _)| *u == uid)
{
Some((name, _, home)) => (uid, name, Some(home)),
None => (uid, format!("uid-{uid}"), None),
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The nursery
// ─────────────────────────────────────────────────────────────────────────────
/// **What a run ended as** — reported per collector, so a service with five healthy
/// collectors and one dead one says exactly that.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Outcome {
/// The collector's `run` returned and the nursery was stopping. Normal.
Stopped,
/// It kept failing and the declared restart budget ran out.
GaveUp { restarts: u32, last: String },
/// It was declared disabled and never started.
Disabled,
}
/// A supervised set of collectors.
pub struct Nursery {
cfg: NurseryConfig,
register: Arc<Register>,
shutdown: Shutdown,
log: LogSink,
/// Total publishes, so a test can assert the register was written rather than
/// inferring it from a row that a single startup publish would also produce.
publishes: Arc<AtomicU64>,
}
impl Nursery {
/// Build a nursery, **refusing a data path that disagrees with the running account**
/// and taking the register's exclusive lock.
pub fn new(cfg: NurseryConfig, log: LogSink) -> Result<Self> {
let (uid, user, home) = running_account();
data_dir_belongs_to_account(&cfg.data_dir, uid, &user, home.as_deref())?;
std::fs::create_dir_all(&cfg.data_dir)
.with_context(|| format!("create the declared data path {}", cfg.data_dir.display()))?;
let writer = Writer::this_process(cfg.substrate.clone(), cfg.unit.clone());
let register = Register::open(&cfg.data_dir, writer)?;
Ok(Self {
cfg,
register: Arc::new(register),
shutdown: Shutdown::new(),
log,
publishes: Arc::new(AtomicU64::new(0)),
})
}
/// Build one over an already-open register — the seam a test uses to keep the writer
/// and reader in one process, and the seam a different substrate uses to nurse a
/// roster it did not declare itself.
pub fn with_register(cfg: NurseryConfig, register: Arc<Register>, log: LogSink) -> Self {
Self {
cfg,
register,
shutdown: Shutdown::new(),
log,
publishes: Arc::new(AtomicU64::new(0)),
}
}
/// The cooperative stop every collector in this nursery observes. Cloned into a signal
/// handler by the runner.
pub fn shutdown(&self) -> Shutdown {
self.shutdown.clone()
}
pub fn register(&self) -> &Arc<Register> {
&self.register
}
pub fn publishes(&self) -> u64 {
self.publishes.load(Ordering::SeqCst)
}
pub fn config(&self) -> &NurseryConfig {
&self.cfg
}
/// **Wait for what the service declared it needs.**
///
/// Both the wait and its verdict sentence come from `skidbladnir::readiness`. There is
/// no probe here: `network-online.target` is passive when both wait-online
/// implementations are masked, and the one place that knows that is the ship.
pub fn await_needs(&self) -> NetworkReady {
match Needs::from(self.cfg.needs) {
Needs::Nothing => NetworkReady::Ready {
waited: Duration::ZERO,
},
Needs::Network => {
let v = readiness::await_network(NETWORK_WAIT);
self.log.line(format!("nursery: {}", v.note()));
v
}
}
}
/// **Run every declared collector until they finish or the nursery is stopped.**
///
/// One tokio runtime, one task per collector. A collector that returns is restarted
/// with backoff until its declared budget runs out; the register carries what it was
/// doing throughout, and what it ended as.
///
/// Returns the outcome per collector. It does not decide what a process should do with
/// that — see [`Nursery::report`], which hands the decision to Skidbladnir.
pub fn run(&self) -> Result<BTreeMap<String, Outcome>> {
let ready = self.await_needs();
if !ready.is_ready() {
// Not fatal by itself: a collector whose source is local may still work, and
// one whose source is not will fail and be restarted — by which time the
// resolver may have arrived. Reporting it is the point.
self.log
.line(format!("nursery: starting anyway — {}", ready.note()));
}
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.context("build the nursery tokio runtime")?;
// **SIGTERM/SIGINT trip the SAME cooperative stop every collector already checks.**
//
// The handler is installed here, on the runtime that owns the collectors, rather
// than in whatever binary called `run` — a second path that stops the same
// collectors is how one of the two ends up unwired. Every supervised run then
// finishes by itself and `run` returns, so a stop is bounded by the collectors and
// never by a detach.
#[cfg(unix)]
{
let sd = self.shutdown.clone();
let log = self.log.clone();
rt.spawn(async move {
use tokio::signal::unix::{signal, SignalKind};
let (mut term, mut int) = match (
signal(SignalKind::terminate()),
signal(SignalKind::interrupt()),
) {
(Ok(t), Ok(i)) => (t, i),
_ => {
log.line(
"nursery: cannot listen for SIGTERM/SIGINT — a stop will have to \
come from the caller's own Shutdown",
);
return;
}
};
tokio::select! {
_ = term.recv() => log.line("nursery: SIGTERM — stopping, waiting for every collector to finish"),
_ = int.recv() => log.line("nursery: SIGINT — stopping, waiting for every collector to finish"),
}
sd.stop();
});
}
let mut planned = Vec::new();
let mut outcomes = BTreeMap::new();
for decl in &self.cfg.collectors {
let Some(c) = daemon_with_config(&decl.name, decl.enabled, decl.config.clone()) else {
bail!(
"no collector named `{}` in this build's registry — `korp-collectors list` \
shows what is compiled in. Refusing to start a partial roster: a service \
that silently skips a declared collector reports healthy while doing less \
than it was told",
decl.name
);
};
// A disabled collector is published once and never run, so an operator asking
// about it gets "off" rather than nothing.
self.publish(c.as_ref())?;
if !decl.enabled {
self.log.line(format!(
"collector {}: declared off — not started",
decl.name
));
outcomes.insert(decl.name.clone(), Outcome::Disabled);
continue;
}
planned.push((decl.clone(), c));
}
if planned.is_empty() {
self.log
.line("nursery: nothing enabled — the roster is declared but every entry is off");
}
self.log.line(format!(
"nursery: {} collector(s) enabled, data {}, register {}. The SINK is in-memory: \
where collected rows land is a separate decision and this service does not make it",
planned.len(),
self.cfg.data_dir.display(),
self.register.path().display()
));
let mut handles = Vec::new();
for (decl, c) in planned {
handles.push((decl.name.clone(), rt.spawn(self.supervise(decl, c))));
}
for (name, h) in handles {
let outcome = rt
.block_on(h)
.with_context(|| format!("join the supervisor task for {name}"))?;
outcomes.insert(name, outcome);
}
Ok(outcomes)
}
/// **Supervise ONE collector: run, publish, restart with backoff.**
///
/// The heartbeat task is what makes the register a live view rather than a transition
/// log: it re-publishes at [`NurseryConfig::heartbeat_ms`] while the run is in flight,
/// so a reader in another process sees the phase move even for a collector that
/// forgets to report one, and can tell a working collector from a stalled one by the
/// row's age.
fn supervise(
&self,
decl: CollectorDecl,
c: Arc<dyn Collector>,
) -> impl std::future::Future<Output = Outcome> + Send + 'static {
let register = Arc::clone(&self.register);
let publishes = Arc::clone(&self.publishes);
let shutdown = self.shutdown.clone();
let log = self.log.clone();
let data_dir = self.cfg.data_dir.clone();
let heartbeat = Duration::from_millis(self.cfg.heartbeat_ms.max(100));
let min = Duration::from_millis(self.cfg.backoff_min_ms.max(1));
let max =
Duration::from_millis(self.cfg.backoff_max_ms.max(self.cfg.backoff_min_ms.max(1)));
async move {
let name = c.name().to_string();
let mut restarts = 0u32;
let mut delay = min;
loop {
if shutdown.is_stopped() {
return Outcome::Stopped;
}
let disk_dir = data_dir.join(&name);
if let Err(e) = std::fs::create_dir_all(&disk_dir) {
let why = format!("cannot create {}: {e}", disk_dir.display());
log.line(format!("collector {name}: {why}"));
let _ = register.retire(
c.as_ref(),
Phase::Failed {
reason: why.clone(),
},
);
return Outcome::GaveUp {
restarts,
last: why,
};
}
let mut ctx = CollectorCtx::in_memory(&name, disk_dir, decl.config.clone());
ctx.fetch = Fetcher::http();
ctx.log = log.clone();
ctx.shutdown = shutdown.clone();
// The phase seam publishes to the register — the existing seam, not a new
// one. Because `Register::publish` samples the collector, a transition
// cannot land in the register without the facet that goes with it.
{
let reg = Arc::clone(®ister);
let pubs = Arc::clone(&publishes);
let watched: Arc<dyn Collector> = Arc::clone(&c);
let nm = name.clone();
let lg = log.clone();
ctx.phase = crate::daemon::PhaseSink::new(move |_p| {
if let Err(e) = reg.publish(watched.as_ref()) {
lg.line(format!("collector {nm}: cannot publish live state: {e:#}"));
} else {
pubs.fetch_add(1, Ordering::SeqCst);
}
});
}
// The heartbeat, stopped by dropping its guard when the run returns.
let beat = {
let reg = Arc::clone(®ister);
let pubs = Arc::clone(&publishes);
let watched: Arc<dyn Collector> = Arc::clone(&c);
tokio::spawn(async move {
loop {
tokio::time::sleep(heartbeat).await;
if reg.publish(watched.as_ref()).is_ok() {
pubs.fetch_add(1, Ordering::SeqCst);
}
}
})
};
let started = Instant::now();
Arc::clone(&c).run(ctx).await;
beat.abort();
let ran = started.elapsed();
// Publish once more, so the register's last word is the state the run
// ended in and not the last transition it happened to announce.
if register.publish(c.as_ref()).is_ok() {
publishes.fetch_add(1, Ordering::SeqCst);
}
if shutdown.is_stopped() {
log.line(format!("collector {name}: stopped after {ran:?}"));
return Outcome::Stopped;
}
let why = match c.phase() {
Phase::Failed { reason } => reason,
other => format!("run returned after {ran:?} in phase `{}`", other.label()),
};
restarts += 1;
if let Some(budget) = decl.max_restarts {
if restarts > budget {
log.line(format!(
"collector {name}: giving up after {restarts} restart(s) — {why}"
));
let _ = register.retire(
c.as_ref(),
Phase::Failed {
reason: format!("gave up after {restarts} restart(s): {why}"),
},
);
return Outcome::GaveUp {
restarts,
last: why,
};
}
}
log.line(format!(
"collector {name}: {why} — restart {restarts} in {delay:?}"
));
// Sleep in slices so a stop during a five-minute backoff is observed.
let until = Instant::now() + delay;
while Instant::now() < until {
if shutdown.is_stopped() {
return Outcome::Stopped;
}
tokio::time::sleep(Duration::from_millis(100).min(until - Instant::now()))
.await;
}
delay = (delay * 2).min(max);
}
}
}
/// Publish one collector's live state.
pub fn publish(&self, c: &dyn Collector) -> Result<()> {
self.register.publish(c)?;
self.publishes.fetch_add(1, Ordering::SeqCst);
Ok(())
}
/// **Hand the outcome to Skidbladnir, which decides what a managed service does when
/// its work dies.**
///
/// This function does not exit and does not choose a code. `report_work_status` is the
/// one place in the house that decides, and it decides the same way for every product
/// — which is the point: a process that keeps listening after its work failed reports
/// `active` to `systemctl` and is left alone for hours.
pub fn report(&self, outcomes: &BTreeMap<String, Outcome>) {
let dead: Vec<String> = outcomes
.iter()
.filter_map(|(n, o)| match o {
Outcome::GaveUp { restarts, last } => {
Some(format!("{n} gave up after {restarts} restart(s): {last}"))
}
_ => None,
})
.collect();
let ran = outcomes.values().any(|o| *o == Outcome::Stopped);
let status = if dead.is_empty() {
WorkStatus::Serving
} else if ran {
// Some collectors are still doing their job. Killing the process would take
// them down with the failure, which is the shared-process weakness this
// service exists to bound — so this is reported, not fatal.
self.log.line(format!(
"nursery: {} of {} collector(s) gave up while others ran — {}",
dead.len(),
outcomes.len(),
dead.join("; ")
));
WorkStatus::Serving
} else {
WorkStatus::Dead(dead.join("; "))
};
readiness::report_work_status("collector-nursery", status);
}
}
// ─────────────────────────────────────────────────────────────────────────────
// RED PROBES
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use crate::register::RegisterReader;
/// ★ **RED — a data path outside the running account's home is REFUSED**, and the
/// refusal names both homes. This is the guard over the measured 1 605 675-tile
/// mismatch, and it must be able to fire.
#[test]
fn a_data_path_under_another_accounts_home_is_refused() {
let err = data_dir_belongs_to_account(
Path::new("/home/korp/.korp/collectors"),
1000,
"alice",
Some(Path::new("/home/alice")),
)
.expect_err("a data path under another account's home must be refused");
let m = err.to_string();
assert!(m.contains("/home/korp/.korp/collectors"), "{m}");
assert!(
m.contains("/home/alice"),
"the refusal must name the home it checked: {m}"
);
assert!(
m.contains("User="),
"the refusal must name where the truth lives: {m}"
);
}
/// …and the matching case passes, or the red above proves nothing.
#[test]
fn a_data_path_under_the_running_accounts_home_is_accepted() {
assert!(
data_dir_belongs_to_account(
Path::new("/home/alice/.collectors/data"),
1000,
"alice",
Some(Path::new("/home/alice")),
)
.is_ok()
);
// root installs into system paths deliberately.
assert!(
data_dir_belongs_to_account(Path::new("/var/lib/collectors"), 0, "root", None).is_ok()
);
}
/// ★ **RED — an account with no passwd home cannot be checked, and that is a refusal,
/// not a pass.** A guard that waves through the case it cannot decide is the shape of
/// a green that can never be red.
#[test]
fn an_account_with_no_passwd_home_is_refused_rather_than_waved_through() {
let err = data_dir_belongs_to_account(Path::new("/srv/data"), 4242, "nobodyish", None)
.expect_err("an undecidable account must refuse");
assert!(err.to_string().contains("no home in passwd"), "{err}");
}
/// ★ **RED — a roster naming a collector this build does not have is REFUSED**, not
/// silently shortened. A service that skips a declared collector reports healthy while
/// doing less than it was told.
#[test]
fn an_unknown_collector_in_the_roster_is_refused_not_skipped() {
let d = tempfile::tempdir().unwrap();
let reg = Arc::new(Register::open(d.path(), Writer::this_process("test", None)).unwrap());
let cfg = NurseryConfig::new(d.path(), vec![CollectorDecl::new("no-such-collector")]);
let n = Nursery::with_register(cfg, reg, LogSink::noop());
let err = n.run().expect_err("an unknown collector must refuse");
assert!(err.to_string().contains("no collector named"), "{err}");
assert!(
err.to_string()
.contains("Refusing to start a partial roster"),
"{err}"
);
}
/// A disabled collector is published as OFF and never started — so an operator asking
/// about it gets an answer, and the register distinguishes "off" from "absent".
#[test]
fn a_disabled_collector_is_published_off_and_not_run() {
let d = tempfile::tempdir().unwrap();
let reg = Arc::new(Register::open(d.path(), Writer::this_process("test", None)).unwrap());
let cfg = NurseryConfig::new(
d.path(),
vec![crate::iceberg_daemons::DcatDaemon::NAME]
.into_iter()
.map(|n| CollectorDecl::new(n).disabled())
.collect(),
);
let n = Nursery::with_register(cfg, Arc::clone(®), LogSink::noop());
let out = n
.run()
.expect("a roster of disabled collectors runs nothing and succeeds");
// Keyed off `DcatDaemon::NAME`, not the literal. The literal is what broke this:
// the config above was already built FROM the const, so the 2026-09-04 rename
// (`dcat` onto its registry twin `dcat-ap-se`) moved one side and left the other,
// and the test read as "the nursery lost a collector" rather than "this string is
// stale". Same shape as the `Step::ALL` expectation that went stale in state.rs.
let dcat = crate::iceberg_daemons::DcatDaemon::NAME;
assert_eq!(out.get(dcat), Some(&Outcome::Disabled));
let v = RegisterReader::open(d.path())
.unwrap()
.unwrap()
.get(dcat)
.unwrap()
.unwrap();
assert!(!v.row.enabled, "a disabled collector's row must say so");
assert_eq!(v.freshness.kind(), "live");
}
/// `Needs::Nothing` must not wait. A nursery that waited for a resolver it never
/// declared needing would add two minutes to every local-only service's boot.
#[test]
fn a_local_only_nursery_does_not_wait_for_the_network() {
let d = tempfile::tempdir().unwrap();
let reg = Arc::new(Register::open(d.path(), Writer::this_process("test", None)).unwrap());
let n = Nursery::with_register(NurseryConfig::new(d.path(), vec![]), reg, LogSink::noop());
let t = Instant::now();
assert!(n.await_needs().is_ready());
assert!(
t.elapsed() < Duration::from_millis(200),
"took {:?}",
t.elapsed()
);
}
/// The declaration round-trips, and a roster read from JSON keeps its defaults.
#[test]
fn the_declaration_round_trips_through_json() {
let cfg =
NurseryConfig::new("/home/x/data", vec![CollectorDecl::new("dcat")]).needing_network();
let back: NurseryConfig =
serde_json::from_str(&serde_json::to_string(&cfg).unwrap()).unwrap();
assert_eq!(back.data_dir, PathBuf::from("/home/x/data"));
assert_eq!(back.needs, NeedsDecl::Network);
assert!(back.collectors[0].enabled, "enabled defaults to true");
// The minimum a foreign project must write.
let minimal: NurseryConfig =
serde_json::from_str(r#"{"data_dir":"/srv/c","collectors":[{"name":"dcat"}]}"#)
.unwrap();
assert_eq!(minimal.collectors.len(), 1);
assert_eq!(minimal.needs, NeedsDecl::Nothing);
assert_eq!(minimal.heartbeat_ms, default_heartbeat_ms());
}
/// ★ **RED — no declaration at all is a refusal.**
#[test]
fn a_nursery_with_no_declaration_refuses_rather_than_inventing_a_roster() {
// Belt and braces: the env var must be absent for this to mean anything.
assert!(
std::env::var(NURSERY_CONFIG_ENV).is_err(),
"this test needs {NURSERY_CONFIG_ENV} unset"
);
let err = NurseryConfig::from_env(None).expect_err("no declaration must refuse");
assert!(
err.to_string().contains("Refusing to invent a roster"),
"{err}"
);
}
}