src/daemon.rs
//! **The collector DAEMON architecture** (feature `daemon`) — the object-safe
//! `Config + State + Collector` trait trio, the `CollectorCtx` a running daemon is
//! handed, and the registry collectors self-register into. Per
//! [`../.nornir/collectors-daemon-design.md`](../.nornir/collectors-daemon-design.md).
//!
//! A **collector** is a *continuous daemon* that strives to reach and hold **ONLINE**
//! for everything in its config, self-healing across restarts — a **state machine**
//! ([`Phase`]), not a one-shot job (that is the older, still-present
//! [`crate::Collector`] ingest trait). `osm-cut` (planet → regions → tiles) is the
//! first implementor ([`crate::osm_cut`], feature `osm-daemon`).
//!
//! ## Leaf discipline
//! korp-collectors depends on **neither korp nor facett** (see `CLAUDE.md`). So the
//! infra the design says a `CollectorCtx` carries — an S3 bucket "via aegir", the
//! "jera daemon-job handle", the log, the BIOS bus — is expressed here as **seams**
//! ([`ObjectStore`], [`PhaseSink`], [`LogSink`], [`BiosBus`]). korp wires the concrete
//! backends (aegir for [`ObjectStore`], a jera [`Reporter`] for [`PhaseSink`]); a test
//! wires the in-memory ones. The leaf pulls nothing new for this feature — it is pure
//! `serde` + `async-trait` + `tokio` (all already present under `parsers`).
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use arrow_array::RecordBatch;
use serde::{Deserialize, Serialize};
// ─────────────────────────────────────────────────────────────────────────────
// Phase — the state machine (BIOS-style pure facts: "not online yet" is normal).
// ─────────────────────────────────────────────────────────────────────────────
/// The collector state machine (design §3). Per target **and** overall:
/// `Offline → Downloading(%) → Indexing → Cutting(region, %) → Online`, with
/// `Degraded`/`Failed{reason}` reachable off any state. Being `Downloading` /
/// "not online yet" is a **normal fact**, never a failure.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum Phase {
/// Nothing acquired yet — the steady pre-work state.
Offline,
/// Acquiring the source (e.g. torrenting the planet); `pct` in `0.0..=1.0`.
Downloading { pct: f32 },
/// Post-acquisition catalog build (planet → region catalog) — no % (fast).
Indexing,
/// Cutting one target's tiles; `pct` in `0.0..=1.0`.
Cutting { region: String, pct: f32 },
/// Steady state: every configured target is served. Green.
Online,
/// Up but impaired (a target failed to refresh, a stale cache, …) — alive,
/// worse than [`Phase::Online`]. Amber, carries the reason.
Degraded { reason: String },
/// Hard failure of the whole collector (bad config, unreadable planet, …). Red.
///
/// **Retryable by nature** — a source that is down, a parse that failed, a disk
/// that filled. The daemon loop backs off and tries again, which is right for all
/// three. It is NOT the state for a missing credential; see [`Phase::Refused`].
Failed { reason: String },
/// **TERMINAL: the collector was never startable, and the reason is NAMED.**
///
/// A declared requirement — a secret, a config field, a source URL — is not
/// satisfied, so there is nothing to attempt. The owner's ruling, 2026-09-06: "a
/// collector whose config is not filled in must be NOT STARTABLE, full stop" — it
/// refuses at load, never half-starts, never retries, never logs a nag.
///
/// **Why this is a separate variant and not a `Failed` with a nicer string.**
/// `Failed` is what the daemon loop backs off and RETRIES. Before this variant
/// existed, a missing `TRAFIKVERKET_API_KEY` was a `Failed`, so the daemon ticked,
/// logged the same reason, doubled its wait from 30 s to 3600 s, and did it again
/// until the process died — nagging an operator hourly about a key that was never
/// coming. The two states need different words because they need different
/// behaviour: one is "try again later", the other is "a human must act, and until
/// they do there is nothing here to run".
///
/// `reason` always names the thing to set (`ENTSOE_TOKEN`, `identitetsbeteckning`)
/// — see [`Missing::line`](crate::requires::Missing::line), which writes it.
Refused { reason: String },
}
impl Default for Phase {
/// The pre-work steady state.
fn default() -> Self {
Phase::Offline
}
}
impl Phase {
/// A short human phase label (the Jobs-view / matrix caption).
pub fn label(&self) -> String {
match self {
Phase::Offline => "offline".into(),
Phase::Downloading { pct } => format!("downloading {:.0}%", pct * 100.0),
Phase::Indexing => "indexing".into(),
Phase::Cutting { region, pct } => format!("cutting {region} {:.0}%", pct * 100.0),
Phase::Online => "online".into(),
Phase::Degraded { reason } => format!("degraded: {reason}"),
Phase::Failed { reason } => format!("failed: {reason}"),
Phase::Refused { reason } => format!("refused: {reason}"),
}
}
/// **Is this a terminal refusal?** — the one question the daemon loop asks before
/// it decides whether to wait and tick again.
pub fn is_refused(&self) -> bool {
matches!(self, Phase::Refused { .. })
}
/// The determinate progress fraction of a working phase (`Downloading` /
/// `Cutting`), or `None` for the phases that have no meaningful %.
pub fn progress(&self) -> Option<f32> {
match self {
Phase::Downloading { pct } | Phase::Cutting { pct, .. } => Some(pct.clamp(0.0, 1.0)),
_ => None,
}
}
/// `true` only in [`Phase::Online`] — the daemon holds its steady state.
pub fn is_online(&self) -> bool {
matches!(self, Phase::Online)
}
/// `true` when the collector is **healthy** — green in the matrix. Only
/// [`Phase::Online`] is healthy; working phases are "not online yet" (a fact, not
/// a failure), and `Degraded`/`Failed` are unhealthy.
pub fn healthy(&self) -> bool {
self.is_online()
}
/// `true` for the working (amber) phases — actively striving toward ONLINE.
pub fn is_working(&self) -> bool {
matches!(
self,
Phase::Downloading { .. } | Phase::Indexing | Phase::Cutting { .. }
)
}
/// The matrix cell colour for this phase given the config `enabled` flag
/// (design §8): green = Online, amber = working / not-yet, grey = off, red =
/// failed. A **disabled** collector is always grey (off) — never red.
pub fn matrix_color(&self, enabled: bool) -> MatrixColor {
if !enabled {
return MatrixColor::Off;
}
match self {
Phase::Online => MatrixColor::Green,
Phase::Failed { .. } => MatrixColor::Red,
// NOT red: nothing is broken. NOT amber: it is not striving toward online
// and never will without a human. NOT off: the operator did not switch it
// off, they never configured it. It gets its own word, the same word korp's
// object registry already uses — `refused`.
Phase::Refused { .. } => MatrixColor::Refused,
// Offline-but-enabled, Downloading, Indexing, Cutting, Degraded are all
// "working / not online yet" — amber, a normal fact, never red.
_ => MatrixColor::Amber,
}
}
}
/// The infra-matrix cell colour (design §8).
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MatrixColor {
/// Online + healthy.
Green,
/// Working — downloading / indexing / cutting / not online yet.
Amber,
/// Disabled in Settings.
Off,
/// Failed.
Red,
/// **Not startable**: a declared requirement (a named secret, a config field, the
/// source URL) is unmet, so the collector refused at load. Distinct from `Off` (a
/// deliberate toggle) and from `Red` (something broke).
Refused,
}
impl MatrixColor {
/// Stable machine key (state_json / tests).
pub fn key(self) -> &'static str {
match self {
MatrixColor::Green => "green",
MatrixColor::Amber => "amber",
MatrixColor::Off => "off",
MatrixColor::Red => "red",
MatrixColor::Refused => "refused",
}
}
/// RGB for the cell (matches korp's collector-matrix legend).
pub fn rgb(self) -> (u8, u8, u8) {
match self {
MatrixColor::Green => (90, 200, 120),
MatrixColor::Amber => (230, 180, 80),
MatrixColor::Off => (120, 120, 120),
MatrixColor::Red => (220, 80, 80),
// Slate blue: legible beside the grey Off it is most easily confused with,
// and deliberately NOT in the red/amber family — nothing is wrong here, a
// key was never supplied.
MatrixColor::Refused => (110, 130, 175),
}
}
/// Is this a "collector is fine" cell (green)?
pub fn is_green(self) -> bool {
matches!(self, MatrixColor::Green)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Config schema — what Settings renders (incl. the region pick-list, design §7).
// ─────────────────────────────────────────────────────────────────────────────
/// One editable config field, described generically so a Settings pane renders it
/// without knowing the collector.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ConfigField {
/// The `custom` JSON key this field reads/writes.
pub key: String,
/// Human label.
pub label: String,
/// Placeholder / help text.
pub hint: String,
/// The widget/type to render.
pub kind: FieldKind,
}
impl ConfigField {
/// A plain text field.
pub fn text(key: &str, label: &str, hint: &str) -> Self {
Self {
key: key.into(),
label: label.into(),
hint: hint.into(),
kind: FieldKind::Text,
}
}
/// A field of `kind`.
pub fn of(key: &str, label: &str, hint: &str, kind: FieldKind) -> Self {
Self {
key: key.into(),
label: label.into(),
hint: hint.into(),
kind,
}
}
}
/// The widget/type of a [`ConfigField`].
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum FieldKind {
/// A boolean toggle.
Bool,
/// A single-line string.
Text,
/// A list of zoom levels (`u8`s), e.g. `[6, 9, 12]`.
Zooms,
/// **The region pick-list** (design §7): a multi-select whose options are the
/// region CATALOG enumerated from the planet. `catalog` is empty before the
/// planet is indexed → Settings shows "planet not downloaded yet" (a fact).
RegionPickList { catalog: Vec<String> },
}
/// The full config schema of a collector — the fields Settings renders (design §1).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ConfigSchema {
pub fields: Vec<ConfigField>,
}
impl ConfigSchema {
pub fn new(fields: Vec<ConfigField>) -> Self {
Self { fields }
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The trait trio (object-safe): Config + State + Collector (the daemon).
// ─────────────────────────────────────────────────────────────────────────────
/// **What you SET** — the config face (design §1). Serde in/out of korp's settings
/// JSON (`collectors.<name> = { enabled, …custom }`). Object-safe.
pub trait CollectorConfig {
/// The stable collector name (the `collectors` map key).
fn name(&self) -> &str;
/// Whether this collector is enabled (from persisted settings).
fn enabled(&self) -> bool;
/// The editable fields — incl. a [`FieldKind::RegionPickList`] whose options are
/// the live region catalog (design §7).
fn schema(&self) -> ConfigSchema;
/// A human title for the ⚙ Settings section / infra-matrix cell. korp renders this
/// generically (registry-driven), so the label travels with the collector, not a
/// korp-side name→title map. Defaults to the bare [`name`](CollectorConfig::name).
fn title(&self) -> String {
self.name().to_string()
}
/// A one-line description under the Settings toggle (what this collector pulls /
/// where it lands). Defaults to empty (no blurb).
fn description(&self) -> String {
String::new()
}
/// **The graph this collector's landed tables honestly make** — the same
/// declaration the ingest trait carries
/// ([`crate::Collector::graph_mappings`]), on the daemon side.
///
/// # Why this had to exist
///
/// korp-collectors has two `Collector` traits: the ingest one (`schema`/`collect`,
/// folded by `registry()`) and this one (`run`, folded by [`daemon_registry`]).
/// Only the ingest trait could declare a graph, and **`osm-cut` implements only
/// this one** — so the collector that lands `graph_nodes.parquet` and
/// `graph_edges.parquet` had nowhere to say what they mean.
///
/// That is the structural reason it grew a direct `GRAPH.BULK` writer instead of a
/// pipeline, and why the road graph has no last run, no freshness and no ▶ Run
/// while every other collector's graph does. A collector that lands tables is a
/// collector, whichever loop it runs in.
///
/// Defaults to none — most daemons land no graph, and saying so is a real answer.
fn graph_mappings(&self) -> Vec<crate::graph_mapping::GraphMapping> {
Vec::new()
}
/// The **default config JSON** — the field defaults Settings seeds into a fresh
/// [`collectors.<name>`] entry when nothing is persisted yet, so a registry-driven
/// Settings pane shows the collector's canonical source URL / table without knowing
/// it. Defaults to an empty object (the schema fields render blank with their hints).
fn defaults(&self) -> serde_json::Value {
serde_json::json!({})
}
/// **The durable working set this collector needs on disk, in bytes** — the ONE
/// place that number is allowed to come from, because the collector is the only
/// thing that knows it. A [`UnitSpec`](crate::runtime::UnitSpec) built for this
/// collector carries it through to whatever the substrate does with it (a PVC
/// request on Kube, a free-space check on metal).
///
/// `None` — the default — means **"undeclared"**, and every consumer must treat it
/// as such. It must NOT be defaulted to a number downstream: korp used to stamp
/// `"20Gi"` onto every unit it built, which read as a declaration and was one for
/// exactly zero collectors. `osm-cut` alone works with a 176 GB planet.
///
/// It does not select a substrate — see [`UnitSpec::placement_advice`](crate::runtime::UnitSpec::placement_advice).
fn durable_bytes(&self) -> Option<u64> {
None
}
/// **What this daemon needs before it can reach ONLINE** — the same declaration the
/// one-shot [`Collector`](crate::Collector) ingest trait answers with
/// ([`crate::requires::Requirements`]): its source URL, its secrets BY NAME, its
/// required config fields.
///
/// The default derives the source from the collector's own
/// [`defaults`](CollectorConfig::defaults) — the `url` / `endpoint` / `source` key a
/// warehouse-sink daemon seeds — and declares no secret. A daemon that reads
/// [`CollectorCtx::secret`] MUST override this and name the variable, because a
/// daemon whose key is absent can only report `Failed` at the wire, which is the
/// state this declaration exists to replace with `NeedsConfig`.
fn requirements(&self) -> crate::requires::Requirements {
let d = self.defaults();
let source = ["url", "endpoint", "source", "base_url"]
.iter()
.find_map(|k| d.get(*k).and_then(|v| v.as_str()))
.unwrap_or_default();
crate::requires::Requirements::open(source)
.inheriting_credentials_of(self.name())
}
}
/// **What it's DOING** — the state face (design §1, §8). The infra matrix reads
/// [`phase`](CollectorState::phase) / [`healthy`](CollectorState::healthy); the
/// click-expand renders [`facet`](CollectorState::facet). Object-safe.
pub trait CollectorState {
/// The current state-machine value (design §3).
fn phase(&self) -> Phase;
/// Green when Online + healthy (design §8). Defaults to the phase's own
/// [`Phase::healthy`]; a collector with per-target health can override.
fn healthy(&self) -> bool {
self.phase().healthy()
}
/// This collector's **own custom UI area** as headless JSON (the matrix
/// click-expand · the state oracle) — the design's `Facet`, expressed as its
/// `state_json` so the leaf needs no `facett` dependency. korp wraps it in a
/// real facett Facet.
fn facet(&self) -> serde_json::Value;
}
/// The matrix cell colour for a collector — its [`Phase`] mapped through the config
/// `enabled` flag (design §8). A free function (not a trait method) so [`Collector`]
/// stays object-safe: `Vec<Arc<dyn Collector>>` calls `matrix_color(&*c)`.
pub fn matrix_color(c: &dyn Collector) -> MatrixColor {
c.phase().matrix_color(c.enabled())
}
/// **The daemon** (design §1, §2) — `Collector : CollectorConfig + CollectorState`.
/// [`run`](Collector::run) is the never-ending loop that ticks the state machine
/// toward ONLINE for every configured target. Object-safe (`self: Arc<Self>` +
/// `#[async_trait]`), so `Vec<Arc<dyn Collector>>` drives the registry.
#[async_trait::async_trait]
pub trait Collector: CollectorConfig + CollectorState + Send + Sync {
/// The continuous daemon loop: reconcile config → do the next needed work →
/// report phase → sleep/watch → repeat. Returns only when
/// [`CollectorCtx::shutdown`] is tripped (the config toggle went off, or the
/// server is winding down) — otherwise it runs forever. It is **restart-safe**:
/// on entry it resumes from what is already on disk / in the bucket.
async fn run(self: Arc<Self>, ctx: CollectorCtx);
}
// ─────────────────────────────────────────────────────────────────────────────
// CollectorCtx — the infra a running daemon is handed (design §4, §5, §6).
// ─────────────────────────────────────────────────────────────────────────────
/// Everything a running collector daemon needs (design §1). Built by korp (aegir
/// bucket + jera reporter + BIOS bus) or by a test (the in-memory seams).
pub struct CollectorCtx {
/// The collector name (== its [`CollectorConfig::name`]).
pub name: String,
/// The per-collector on-disk dir — `~/.korp/collectors/<name>/` (design §4): the
/// planet `.pbf`, the region catalog, random-access caches, batched intermediates.
pub disk_dir: PathBuf,
/// The persisted custom config (`collectors.<name>` minus `enabled`).
pub config: serde_json::Value,
/// The S3 bucket seam — servable artifacts (`nordisk-maps/<region>/tiles/…`).
/// korp backs this with aegir; a test with [`MemObjectStore`]. The S3-sink
/// collectors (`osm-cut`) write here; the Iceberg-sink collectors ignore it.
pub bucket: Arc<dyn ObjectStore>,
/// The **Iceberg / warehouse sink** seam (design §11) — where the *warehouse-sink*
/// collectors (`dcat`, `pxweb`, `osm-admin`) land their parsed Arrow
/// `RecordBatch`es, one per configured table. korp backs this with a skade
/// `Table` (`BatchSink::iceberg`); a test with [`MemWarehouse`]. This is the seam
/// that makes the daemon architecture generic over BOTH sink kinds (an S3 bucket
/// AND an Iceberg table) — the only thing an S3-sink and a warehouse-sink daemon
/// differ in is which of these two they use.
pub warehouse: Arc<dyn WarehouseSink>,
/// The **external-source fetch** seam — how a daemon acquires its source bytes
/// (the nightly RDF dump, a PxWeb JSON-stat2 cube, an Overpass admin-boundary
/// query). korp backs it with the shared rustls HTTP core; a test with a canned
/// [`Fetcher::from_map`] fixture, so the run-loop + sink + matrix wiring is proven
/// with **no live gov API** (the offline-safe contract). `osm-cut` resolves its
/// planet from disk and does not use this.
pub fetch: Fetcher,
/// The jera daemon-job handle seam — where the live phase/%/healthy is pushed
/// (design §5). korp wires this to a jera `Reporter`.
pub phase: PhaseSink,
/// The normal, detailed work log (design §6) — "planet 42%", "cut sweden → 47831
/// tiles", warnings.
pub log: LogSink,
/// The coarse, append-only **BIOS status bus** (design §6) — pure facts the
/// server aggregates at startup ("collector map: not online yet → ONLINE").
pub bios: BiosBus,
/// Cooperative stop — tripped when the config toggle goes off or the server
/// shuts down. The daemon loop checks [`Shutdown::is_stopped`] between ticks.
pub shutdown: Shutdown,
/// The **secret-resolution seam** — where a daemon's API key / token / password comes
/// from (design: no daemon calls `std::env::var` directly). The leaf defaults it to
/// [`EnvSecretStore`] (env var / inline literal); korp injects a vault-backed
/// [`SecretStore`] here WITHOUT this crate gaining a vault dependency. A daemon that
/// needs a secret asks this and reports `Failed` when it answers `None`.
pub secret: Arc<dyn SecretStore>,
/// **The read-watermark seam** — where a daemon's incremental tick reads and
/// advances the mark for `(collector, source)`.
///
/// korp wires the SAME store the ▶ Collect button uses
/// (`~/.korp/collector_marks.json`), so a manual collect and an unattended tick
/// share ONE watermark and neither re-reads what the other already took. A test
/// gets [`MemMarkStore`](crate::freshness::MemMarkStore), whose honest behaviour is
/// that every run is a first run — safe (it re-reads) rather than wrong (it skips).
pub marks: Arc<dyn crate::freshness::MarkStore>,
/// **The wake-up seam** — a host-side event (an invalidated step, a changed
/// region config) [`Nudge::poke`]s this, and the daemon's idle wait cuts short
/// and reconciles NOW instead of at the next refresh tick. Without it a purged
/// step waits up to `refresh_secs` (default 3600 s) before anything notices.
pub nudge: Nudge,
/// **The per-step job-history seam** — where a daemon records one durable row
/// per unit of work that actually RAN (a convert pass, a chain step, a planet
/// acquisition), so the per-leg history the boot provisioner used to write does
/// not disappear when the daemon owns the chain. korp backs this with jera rows
/// carrying `parent_id` = the daemon's own job row; a test (and the leaf default)
/// gets [`ChildJobs::noop`]. Ticks that find a step `AlreadyComplete` record
/// nothing — the ledger holds work, not confirmations.
pub jobs: Arc<dyn ChildJobs>,
}
impl CollectorCtx {
/// A ctx with the in-memory seams — a self-contained daemon for tests/benches.
/// `disk_dir` is where the daemon stages its planet/catalog/tiles.
pub fn in_memory(name: &str, disk_dir: PathBuf, config: serde_json::Value) -> Self {
Self {
name: name.into(),
disk_dir,
config,
bucket: Arc::new(MemObjectStore::new()),
warehouse: Arc::new(MemWarehouse::new()),
fetch: Fetcher::noop(),
phase: PhaseSink::noop(),
log: LogSink::noop(),
bios: BiosBus::noop(),
shutdown: Shutdown::new(),
secret: std::sync::Arc::new(crate::daemon::EnvSecretStore::new()),
marks: std::sync::Arc::new(crate::freshness::MemMarkStore::new()),
nudge: Nudge::new(),
jobs: <dyn ChildJobs>::noop(),
}
}
/// Report a new phase everywhere it should go: the jera handle ([`PhaseSink`])
/// AND a matching BIOS fact (ONLINE vs "not online yet"). The one call a daemon
/// makes on every state transition.
pub fn set_phase(&self, phase: &Phase) {
self.phase.set(phase);
if phase.is_online() {
self.bios.online(&self.name);
} else if let Phase::Failed { reason } = phase {
self.bios
.post(&self.name, &format!("FAILED: {reason}"), false);
} else if let Phase::Refused { reason } = phase {
// A refusal is a FACT for the BIOS bus like any other, and it is posted
// ONCE because the daemon reaches it once and stops. `REFUSED:` rather than
// `FAILED:` so an operator reading the boot log can tell "give me a key"
// from "something is broken".
self.bios
.post(&self.name, &format!("REFUSED: {reason}"), false);
} else {
self.bios.not_online_yet(&self.name, &phase.label());
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// SecretStore seam — RE-EXPORTED from `crate::requires`.
//
// The seam moved there (unchanged) when readiness learned to ask it whether a named
// secret is present: readiness answers for the one-shot ingest registry too, which is
// compiled WITHOUT the `daemon` feature, and a second copy of `SecretStore` behind a
// second gate would be two answers to "is this key set?". These aliases keep every
// existing `korp_collectors::daemon::…` path working.
// ─────────────────────────────────────────────────────────────────────────────
pub use crate::requires::{EnvSecretStore, SecretStore, SecretString};
// ─────────────────────────────────────────────────────────────────────────────
// ObjectStore seam — the S3 bucket (design §4). aegir-backed in korp.
// ─────────────────────────────────────────────────────────────────────────────
/// The servable-artifact store (design §4) — an S3 bucket abstracted so the leaf
/// needs no aegir dependency. korp implements this over aegir's SigV4 client; a
/// test over [`MemObjectStore`].
pub trait ObjectStore: Send + Sync {
/// Create the bucket if it does not exist (idempotent). Default: no-op (a store
/// with no bucket concept).
fn ensure_bucket(&self, _bucket: &str) -> anyhow::Result<()> {
Ok(())
}
/// Upload `bytes` under `bucket/key`.
fn put(&self, bucket: &str, key: &str, bytes: &[u8]) -> anyhow::Result<()>;
/// Fetch `bucket/key`, or `None` if absent.
fn get(&self, bucket: &str, key: &str) -> anyhow::Result<Option<Vec<u8>>>;
/// List every key under `bucket` starting with `prefix`.
fn list(&self, bucket: &str, prefix: &str) -> anyhow::Result<Vec<String>>;
/// **Retire `bucket/key`.** Deleting an absent key is `Ok` (idempotent), so a
/// retire loop can be re-run; a transport/permission failure is an `Err` the
/// caller must report rather than swallow.
///
/// # Why this is a REQUIRED method, not a defaulted no-op
/// Until 2026-08-02 this seam had only `put` + `list`, so
/// [`crate::osm_cut::cut_region_to_store`] **structurally could not** retire a
/// stale key: a re-cut that legitimately produces FEWER tiles (which is exactly
/// what bounding the pyramid to the region does) left every wider run's key
/// standing, and the reader fetches by KEY — so every stray object kept
/// answering 200 and the fix read as a no-op on any box that already had a
/// scattered pyramid. MEASURED that day: korp's own one-shot
/// (`korp::map_tiles::run_tile_maps`) retired **264** foreign-only objects from
/// `s3://nordisk-maps/sweden/tiles/`; the collector — the documented successor —
/// could not have retired one of them.
///
/// A defaulted `Ok(())` would have been worse than the missing method: it would
/// let an implementor report a clean retire while deleting nothing. An
/// implementor that genuinely cannot delete must say so with an `Err`.
fn delete(&self, bucket: &str, key: &str) -> anyhow::Result<()>;
}
/// An in-memory [`ObjectStore`] — the default in [`CollectorCtx::in_memory`] and the
/// store daemon tests/benches write tiles into. Cheap to clone (shares the map).
/// `(bucket, key) → bytes` — the [`MemObjectStore`] backing map.
type ObjectMap = Arc<Mutex<BTreeMap<(String, String), Vec<u8>>>>;
#[derive(Clone, Default)]
pub struct MemObjectStore {
objects: ObjectMap,
}
impl MemObjectStore {
pub fn new() -> Self {
Self::default()
}
/// How many objects are stored under `bucket` (all prefixes) — test convenience.
pub fn count(&self, bucket: &str) -> usize {
self.objects
.lock()
.unwrap_or_else(|e| e.into_inner())
.keys()
.filter(|(b, _)| b == bucket)
.count()
}
}
impl ObjectStore for MemObjectStore {
fn put(&self, bucket: &str, key: &str, bytes: &[u8]) -> anyhow::Result<()> {
self.objects
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert((bucket.to_string(), key.to_string()), bytes.to_vec());
Ok(())
}
fn get(&self, bucket: &str, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
Ok(self
.objects
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(&(bucket.to_string(), key.to_string()))
.cloned())
}
fn list(&self, bucket: &str, prefix: &str) -> anyhow::Result<Vec<String>> {
Ok(self
.objects
.lock()
.unwrap_or_else(|e| e.into_inner())
.keys()
.filter(|(b, k)| b == bucket && k.starts_with(prefix))
.map(|(_, k)| k.clone())
.collect())
}
fn delete(&self, bucket: &str, key: &str) -> anyhow::Result<()> {
self.objects
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&(bucket.to_string(), key.to_string()));
Ok(())
}
}
// ─────────────────────────────────────────────────────────────────────────────
// WarehouseSink seam — the Iceberg table sink (design §11). skade-backed in korp.
// ─────────────────────────────────────────────────────────────────────────────
/// The **Iceberg / warehouse table sink** (design §11) — the counterpart of
/// [`ObjectStore`] for the *warehouse-sink* collectors (`dcat`, `pxweb`,
/// `osm-admin`). A daemon appends parsed Arrow [`RecordBatch`]es to a named table;
/// korp backs this with a skade `Table` (`korp_collectors::sink::BatchSink::iceberg`
/// → `fast_append`), a test with [`MemWarehouse`]. Object-safe so
/// `CollectorCtx.warehouse` is an `Arc<dyn WarehouseSink>`.
pub trait WarehouseSink: Send + Sync {
/// Append `batch` to `table`, landing it in the warehouse. Returns the number of
/// rows appended. An empty batch is a no-op (0). skade recasts each batch to the
/// table schema; a memory sink just buffers it.
fn append(&self, table: &str, batch: &RecordBatch) -> anyhow::Result<u64>;
/// Rows landed in `table` so far — the daemon reads this for its facet / the
/// "empty-is-RED" check, and tests assert it. Best-effort for a live warehouse
/// (a memory sink is exact; a skade sink reports what it has committed).
fn rows(&self, table: &str) -> u64;
}
/// `table → appended batches` — the [`MemWarehouse`] backing map.
type WarehouseMap = Arc<Mutex<BTreeMap<String, Vec<RecordBatch>>>>;
/// An in-memory [`WarehouseSink`] — the default in [`CollectorCtx::in_memory`] and
/// what the Iceberg-sink daemon tests/benches land rows into. Cheap to clone (shares
/// the map).
#[derive(Clone, Default)]
pub struct MemWarehouse {
tables: WarehouseMap,
}
impl MemWarehouse {
pub fn new() -> Self {
Self::default()
}
/// The batches appended to `table` (test convenience — read the Arrow back).
pub fn batches(&self, table: &str) -> Vec<RecordBatch> {
self.tables
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(table)
.cloned()
.unwrap_or_default()
}
/// The table names written so far.
pub fn tables(&self) -> Vec<String> {
self.tables
.lock()
.unwrap_or_else(|e| e.into_inner())
.keys()
.cloned()
.collect()
}
}
impl WarehouseSink for MemWarehouse {
fn append(&self, table: &str, batch: &RecordBatch) -> anyhow::Result<u64> {
let n = batch.num_rows() as u64;
if n == 0 {
return Ok(0);
}
self.tables
.lock()
.unwrap_or_else(|e| e.into_inner())
.entry(table.to_string())
.or_default()
.push(batch.clone());
Ok(n)
}
fn rows(&self, table: &str) -> u64 {
self.tables
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(table)
.map(|bs| bs.iter().map(|b| b.num_rows() as u64).sum())
.unwrap_or(0)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Fetcher seam — how a daemon acquires its external source bytes (design §11).
// ─────────────────────────────────────────────────────────────────────────────
/// How a warehouse-sink daemon **fetches its external source** — a `Fn(url) → bytes`
/// GET closure PLUS a `Fn(url, body, content-type) → bytes` POST closure, so a source
/// that answers a query POSTed in the request body (Trafikverket's XML `<REQUEST>`)
/// rides the same seam as the plain-GET sources. The leaf pulls no HTTP client into
/// `daemon` (korp wires the shared rustls core; a test wires a canned
/// [`from_map`](Fetcher::from_map) fixture — which serves BOTH verbs by URL, so the
/// run-loop is proven with no live gov API). Cheap to clone.
#[derive(Clone)]
pub struct Fetcher {
get: Arc<dyn Fn(&str) -> anyhow::Result<Vec<u8>> + Send + Sync>,
post: Arc<dyn Fn(&str, &str, &str) -> anyhow::Result<Vec<u8>> + Send + Sync>,
/// **The same fixture, in the shape `collect()` can read.**
///
/// A daemon tick routed through its registry twin does not call these closures at
/// all — [`Collector::collect`](crate::Collector::collect) builds its own HTTP core
/// from [`CollectorCfg`](crate::CollectorCfg). So the offline-ness of a fetcher has
/// to be a property the *cfg* can inherit, or a test wires one seam and the other
/// silently goes live (measured: a canned polisen tick landed 500 live rows).
///
/// Set by [`from_map`](Fetcher::from_map) — the fixture constructor — and `None`
/// for every live one, so a canned fetcher gives a canned collect and a live
/// fetcher gives a live collect, with nothing to remember.
canned: Option<crate::http::Canned>,
}
impl Fetcher {
/// A GET-only fetcher that hands each `url` to `f` (korp's HTTP-core bridge). Its
/// POST verb is unwired and errors loudly — a caller that needs POST builds one of
/// the verb-aware constructors ([`http`](Fetcher::http) / [`from_map`](Fetcher::from_map)).
pub fn new(f: impl Fn(&str) -> anyhow::Result<Vec<u8>> + Send + Sync + 'static) -> Self {
Fetcher {
get: Arc::new(f),
post: Arc::new(|url, _body, _ct| {
anyhow::bail!("this fetcher is GET-only (cannot POST {url})")
}),
canned: None,
}
}
/// A fetcher with no source wired — every fetch errors. The default in
/// [`CollectorCtx::in_memory`] (an S3-sink daemon never fetches through it).
pub fn noop() -> Self {
Fetcher {
get: Arc::new(|url| anyhow::bail!("no fetcher wired (cannot fetch {url})")),
post: Arc::new(|url, _b, _c| anyhow::bail!("no fetcher wired (cannot fetch {url})")),
canned: None,
}
}
/// A canned fetcher: `url → bytes`, the hermetic test/bench fixture (no network).
/// An un-mapped url errors (a mis-wired daemon fails loudly, never silently empty).
/// A POST resolves by URL too (the request body is ignored) so a POST-body source
/// (Trafikverket) is exercised offline against the same fixture map.
pub fn from_map(map: BTreeMap<String, Vec<u8>>) -> Self {
let canned = Some(crate::http::Canned::new(map.clone()));
let map = Arc::new(map);
let get_map = map.clone();
Fetcher {
canned,
get: Arc::new(move |url: &str| {
get_map
.get(url)
.cloned()
.ok_or_else(|| anyhow::anyhow!("no fixture bytes for {url}"))
}),
post: Arc::new(move |url: &str, _body: &str, _ct: &str| {
map.get(url)
.cloned()
.ok_or_else(|| anyhow::anyhow!("no fixture bytes for {url}"))
}),
}
}
/// A **live HTTP fetcher** over the shared rustls stack (blocking — a daemon calls
/// [`get`](Fetcher::get)/[`post`](Fetcher::post) from a `spawn_blocking` worker, so a
/// synchronous request is correct and never blocks the reactor). A non-2xx or transport
/// error surfaces as a loud `Err` (never a silent empty). This is what korp wires behind
/// the seam for the shipped warehouse-sink daemons; a test wires
/// [`from_map`](Fetcher::from_map). Built once and shared across the daemon's periodic
/// re-fetches.
pub fn http() -> Self {
Fetcher {
canned: None,
get: Arc::new(move |url: &str| {
let client = shared_blocking_client();
let resp = client
.get(url)
.send()
.map_err(|e| anyhow::anyhow!("GET {url}: {e}"))?;
let status = resp.status();
if !status.is_success() {
anyhow::bail!("GET {url}: HTTP {status}");
}
let bytes = resp
.bytes()
.map_err(|e| anyhow::anyhow!("read body {url}: {e}"))?;
Ok(bytes.to_vec())
}),
post: Arc::new(move |url: &str, body: &str, ct: &str| {
let client = shared_blocking_client();
let resp = client
.post(url)
.header("content-type", ct)
.body(body.to_string())
.send()
.map_err(|e| anyhow::anyhow!("POST {url}: {e}"))?;
let status = resp.status();
if !status.is_success() {
anyhow::bail!("POST {url}: HTTP {status}");
}
let bytes = resp
.bytes()
.map_err(|e| anyhow::anyhow!("read body {url}: {e}"))?;
Ok(bytes.to_vec())
}),
}
}
/// A fetcher with EXPLICIT get + post closures — the seam a test uses to CAPTURE the
/// request body (proving the resolved secret reaches the wire) while returning a canned
/// reply, or that a host uses to wire a bespoke transport for both verbs.
pub fn from_verbs(
get: impl Fn(&str) -> anyhow::Result<Vec<u8>> + Send + Sync + 'static,
post: impl Fn(&str, &str, &str) -> anyhow::Result<Vec<u8>> + Send + Sync + 'static,
) -> Self {
Fetcher {
get: Arc::new(get),
post: Arc::new(post),
canned: None,
}
}
/// **The fixture this fetcher is backed by, if it is a fixture at all.**
///
/// `Some` only for [`from_map`](Fetcher::from_map). A daemon tick that routes
/// through its registry twin hands this to the twin's
/// [`CollectorCfg`](crate::CollectorCfg), so ONE declaration makes both the
/// leaf-side fetch and the `collect()`-side fetch offline. `None` means live, and a
/// live fetcher must give a live collect — anything else would be a test seam
/// deciding production behaviour.
pub fn canned(&self) -> Option<crate::http::Canned> {
self.canned.clone()
}
/// Fetch `url` → bytes (GET).
pub fn get(&self, url: &str) -> anyhow::Result<Vec<u8>> {
(self.get)(url)
}
/// POST `body` (with `content_type`) to `url` → bytes — the query-in-the-body verb.
pub fn post(&self, url: &str, body: &str, content_type: &str) -> anyhow::Result<Vec<u8>> {
(self.post)(url, body, content_type)
}
}
/// **The process's ONE blocking HTTP client, built off any async context and never
/// dropped.**
///
/// `reqwest::blocking::Client` owns an internal tokio runtime. Building or dropping one
/// while an async context is current panics — *"Cannot drop a runtime in a context where
/// blocking is not allowed"* — and that is not theoretical: a supervisor that set
/// `ctx.fetch = Fetcher::http()` and then handed the ctx to `run` took the whole collector
/// down on the drop, before the collector had fetched anything.
///
/// So the client is built exactly once, **on a plain thread with no runtime installed**,
/// and lives in a `OnceLock` for the life of the process. Nothing ever drops it, so the
/// hazardous drop cannot happen from anywhere; and a `Fetcher` becomes a cheap handle a
/// supervisor may build inside an async block without thinking about it.
fn shared_blocking_client() -> &'static reqwest::blocking::Client {
static CLIENT: std::sync::OnceLock<reqwest::blocking::Client> = std::sync::OnceLock::new();
CLIENT.get_or_init(|| {
build_off_runtime(|| {
tls_roots_blocking(reqwest::blocking::Client::builder())
.timeout(std::time::Duration::from_secs(60))
.user_agent("korp-collectors-daemon")
.build()
.expect("build blocking rustls http client")
})
})
}
/// **Run `f` on a fresh thread, so no caller's ambient tokio runtime is current while it
/// runs.**
///
/// The one place that trick lives. `reqwest::blocking` starts a runtime of its own on
/// build, and doing that under someone else's runtime is the first half of the panic pair
/// (the second half is the drop, which a `OnceLock` prevents by never dropping). Every
/// blocking client in this crate is built through here, so there is one thread hop to
/// reason about rather than one per call site.
pub(crate) fn build_off_runtime<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> T {
std::thread::spawn(f)
.join()
.expect("the off-runtime builder thread panicked")
}
// ─────────────────────────────────────────────────────────────────────────────
// PhaseSink seam — the jera daemon-job handle (design §5).
// ─────────────────────────────────────────────────────────────────────────────
/// Where a daemon pushes its live [`Phase`] — the jera daemon-job handle seam
/// (design §5). korp wires this to a jera `Reporter` (`reporter.phase(label,
/// healthy)` + `reporter.progress(...)`); a test to a recorder. Cheap to clone.
#[derive(Clone)]
pub struct PhaseSink(Arc<dyn Fn(&Phase) + Send + Sync>);
impl PhaseSink {
/// A sink that hands each phase to `f` (korp's jera-reporter bridge).
pub fn new(f: impl Fn(&Phase) + Send + Sync + 'static) -> Self {
PhaseSink(Arc::new(f))
}
/// A sink that drops every phase (unwired).
pub fn noop() -> Self {
PhaseSink(Arc::new(|_| {}))
}
/// Push a phase.
pub fn set(&self, phase: &Phase) {
(self.0)(phase)
}
/// A recorder + its shared log — for tests that assert the phase sequence.
pub fn recorder() -> (Self, Arc<Mutex<Vec<Phase>>>) {
let log: Arc<Mutex<Vec<Phase>>> = Arc::new(Mutex::new(Vec::new()));
let l2 = log.clone();
(
PhaseSink(Arc::new(move |p: &Phase| {
l2.lock().unwrap_or_else(|e| e.into_inner()).push(p.clone())
})),
log,
)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// LogSink seam — the normal work log (design §6).
// ─────────────────────────────────────────────────────────────────────────────
/// The detailed per-collector work log (design §6). korp routes this to its normal
/// log; a bare daemon to stderr; a test to nowhere. Cheap to clone.
#[derive(Clone)]
pub struct LogSink(Arc<dyn Fn(&str) + Send + Sync>);
impl LogSink {
pub fn new(f: impl Fn(&str) + Send + Sync + 'static) -> Self {
LogSink(Arc::new(f))
}
/// Log to stderr (prefixed) — a bare daemon's default.
pub fn stderr() -> Self {
LogSink(Arc::new(|m: &str| eprintln!("collector: {m}")))
}
pub fn noop() -> Self {
LogSink(Arc::new(|_| {}))
}
/// Emit one log line.
pub fn line(&self, msg: impl AsRef<str>) {
(self.0)(msg.as_ref())
}
/// A recorder + its shared lines — for tests that assert what a pass DID (e.g.
/// that it never went near the planet). Mirrors [`PhaseSink::recorder`].
pub fn recorder() -> (Self, Arc<Mutex<Vec<String>>>) {
let lines: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let l2 = lines.clone();
(
LogSink(Arc::new(move |m: &str| {
l2.lock()
.unwrap_or_else(|e| e.into_inner())
.push(m.to_string())
})),
lines,
)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// BIOS status bus (design §6) — pure-fact, append-only, non-blocking.
// ─────────────────────────────────────────────────────────────────────────────
/// One coarse, pure fact on the BIOS status bus (design §6). Append-only; the
/// server aggregates these at startup like a hardware BIOS POST — reporting facts
/// as each collector "tunes in" without ever blocking or invalidating startup.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct BiosFact {
/// The collector this fact is about.
pub collector: String,
/// The pure-fact message ("not online yet (downloading 12%)", "ONLINE", …).
pub message: String,
/// Whether this fact means the collector is now ONLINE.
pub online: bool,
/// A monotonic sequence (append order) — not wall-clock, so ordering is exact.
pub seq: u64,
}
/// The **poster** side of the BIOS bus a [`CollectorCtx`] carries (design §6). Cheap
/// to clone. Non-blocking: posting a fact never waits on a reader. korp backs it with
/// a [`BiosLog`]; a bare daemon with [`BiosBus::noop`].
/// The BIOS-fact sink closure: `(collector, message, online, seq)`.
type BiosSink = Arc<dyn Fn(&str, &str, bool, u64) + Send + Sync>;
#[derive(Clone)]
pub struct BiosBus {
sink: BiosSink,
seq: Arc<AtomicU64>,
}
impl BiosBus {
/// A bus that hands each fact to `f`.
pub fn new(f: impl Fn(&str, &str, bool, u64) + Send + Sync + 'static) -> Self {
BiosBus {
sink: Arc::new(f),
seq: Arc::new(AtomicU64::new(0)),
}
}
/// A bus that drops every fact.
pub fn noop() -> Self {
BiosBus::new(|_, _, _, _| {})
}
/// Post a raw fact.
pub fn post(&self, collector: &str, message: &str, online: bool) {
let seq = self.seq.fetch_add(1, Ordering::Relaxed);
(self.sink)(collector, message, online, seq);
}
/// "collector X: not online yet (<detail>)" — the normal, non-failing fact.
pub fn not_online_yet(&self, collector: &str, detail: &str) {
self.post(collector, &format!("not online yet ({detail})"), false);
}
/// "collector X: ONLINE" — the tuned-in fact.
pub fn online(&self, collector: &str) {
self.post(collector, "ONLINE", true);
}
/// Post the **build stamp** — the crate version + git hash
/// ([`crate::version_string`]) — as a BIOS fact, so the startup banner and the
/// health surface record exactly which build is running. Not an online/offline
/// signal (`online = false`), a pure provenance fact.
pub fn version(&self, collector: &str) {
self.post(collector, &crate::version_string(), false);
}
}
/// The server-side **aggregator** of the BIOS bus (design §6, §8) — the append-only
/// fact log korp renders as a startup banner and reads to colour the matrix / decide
/// "is collector X online yet". `Arc`-shared so a [`BiosBus`] posts into it from a
/// daemon thread while the UI reads it per-frame. Never blocks startup.
#[derive(Clone, Default)]
pub struct BiosLog {
facts: Arc<Mutex<Vec<BiosFact>>>,
}
impl BiosLog {
pub fn new() -> Self {
Self::default()
}
/// A [`BiosBus`] that appends into this log.
pub fn bus(&self) -> BiosBus {
let facts = self.facts.clone();
BiosBus::new(move |collector, message, online, seq| {
facts
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(BiosFact {
collector: collector.to_string(),
message: message.to_string(),
online,
seq,
});
})
}
/// Every fact posted so far, in append order.
pub fn snapshot(&self) -> Vec<BiosFact> {
self.facts.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// Whether `collector` has posted an ONLINE fact (its latest online-bearing
/// fact is online). "Not online yet" facts never flip this true.
pub fn is_online(&self, collector: &str) -> bool {
self.facts
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.rfind(|f| f.collector == collector)
.map(|f| f.online)
.unwrap_or(false)
}
/// Render the append-only bus as a BIOS-POST-style banner (one line per fact).
pub fn render(&self) -> String {
self.snapshot()
.iter()
.map(|f| format!("collector {}: {}", f.collector, f.message))
.collect::<Vec<_>>()
.join("\n")
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Shutdown — cooperative stop for the daemon loop.
// ─────────────────────────────────────────────────────────────────────────────
/// A cooperative stop flag the daemon loop checks between ticks — tripped when the
/// config toggle goes off (stoppable, design §5) or the server shuts down. Cheap to
/// clone; all clones share one flag.
#[derive(Clone, Debug, Default)]
pub struct Shutdown(Arc<AtomicBool>);
impl Shutdown {
pub fn new() -> Self {
Self::default()
}
/// Request the daemon stop at its next tick.
pub fn stop(&self) {
self.0.store(true, Ordering::SeqCst);
}
/// Has a stop been requested?
pub fn is_stopped(&self) -> bool {
self.0.load(Ordering::SeqCst)
}
}
/// **The wake-up flag** ([`CollectorCtx::nudge`]) — a host-side event poked this
/// daemon, and its idle wait should end now. One flag, not a queue: two pokes
/// before the daemon wakes mean ONE reconcile, which is right — the reconcile
/// reads disk truth, not the events.
#[derive(Clone, Default)]
pub struct Nudge(Arc<AtomicBool>);
impl Nudge {
pub fn new() -> Self {
Self::default()
}
/// Wake the daemon: its current idle wait ends and it reconciles.
pub fn poke(&self) {
self.0.store(true, Ordering::SeqCst);
}
/// Consume the poke (the daemon's wait loop): `true` once per poke-set.
pub fn take(&self) -> bool {
self.0.swap(false, Ordering::SeqCst)
}
}
/// **The per-step job-history seam** ([`CollectorCtx::jobs`]) — one durable row per
/// unit of work a daemon actually RAN. korp backs it with jera rows whose
/// `parent_id` is the daemon's own job row; the leaf default records nothing.
/// One post-fact call, not a begin/finish pair: the daemon already measured the
/// step, and a handle that must be carried across an await point would force the
/// seam into every signature it crosses.
pub trait ChildJobs: Send + Sync {
/// Record one finished unit of work: `target` names it (`"sweden:e"`,
/// `"planet"`), `ok` is its verdict, `wall_ms` its measured cost, `detail`
/// whatever the step itself said (rows, bytes, the error).
fn record(&self, target: &str, ok: bool, wall_ms: u64, detail: serde_json::Value);
}
/// The leaf default: no job history. A real sink is the host's to inject.
pub struct NoopChildJobs;
impl ChildJobs for NoopChildJobs {
fn record(&self, _target: &str, _ok: bool, _wall_ms: u64, _detail: serde_json::Value) {}
}
impl dyn ChildJobs {
/// The shared no-op sink for tests and leaf-only builds.
pub fn noop() -> Arc<dyn ChildJobs> {
Arc::new(NoopChildJobs)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Registry — collectors self-register (design §1).
// ─────────────────────────────────────────────────────────────────────────────
// ── The EXTERNAL registry — collectors that live in another crate ────────────
/// How a foreign crate builds one of its collectors: `(enabled, config) -> collector`.
/// The same signature [`daemon_with_config`] uses for the built-ins, so the two
/// registries answer the same question the same way.
type ExternalFactory = Arc<dyn Fn(bool, serde_json::Value) -> Arc<dyn Collector> + Send + Sync>;
fn external_registry() -> &'static Mutex<BTreeMap<String, ExternalFactory>> {
static REG: std::sync::OnceLock<Mutex<BTreeMap<String, ExternalFactory>>> =
std::sync::OnceLock::new();
REG.get_or_init(|| Mutex::new(BTreeMap::new()))
}
/// **Register a collector that this crate does not contain.**
///
/// # Why this exists
///
/// [`crate::nursery`]'s stated purpose is a foreign project installing the nursery and
/// *"lägga in sina egna collectorer utan att hela OpenStreetMap dras med"*
/// (`nursery.rs:6-8`). It could not do that: [`Nursery::run`](crate::nursery::Nursery::run)
/// resolves every declared name through [`daemon_with_config`], which was a closed
/// if-chain over this crate's own collectors — so a foreign collector's only possible
/// outcome was the *"no collector named …"* refusal, which is the right answer to a typo
/// and the wrong answer to an extension. The nursery's own doc promised a door that had
/// no handle on the outside.
///
/// # The rules, and why each one is a refusal rather than a warning
///
/// * **A name already taken is REFUSED**, whether by a built-in or by an earlier
/// registration. Last-write-wins would let a foreign crate silently shadow `dcat`, and
/// the operator's roster would name one collector while running another.
/// * **Registration is process-wide and permanent.** There is no deregister: a nursery
/// that could lose a collector mid-run would report a roster it is no longer running.
/// * Call it **before** [`Nursery::run`](crate::nursery::Nursery::run) — from `main`, or
/// from the consumer's own registration function. A name that arrives after the roster
/// is resolved is a name the roster never saw.
pub fn register_external_daemon(
name: impl Into<String>,
factory: impl Fn(bool, serde_json::Value) -> Arc<dyn Collector> + Send + Sync + 'static,
) -> anyhow::Result<()> {
let name = name.into();
if name.trim().is_empty() {
anyhow::bail!("a collector must have a name — an empty one can never be declared");
}
// Asked in this order so each refusal names the RIGHT collision: a second foreign
// registration and a clash with a built-in are different mistakes with different
// fixes, and one message covering both would send the reader to the wrong crate.
if external_daemon_names().contains(&name) {
anyhow::bail!(
"`{name}` is already registered by another crate in this process. Refusing to \
replace it: an operator's roster would name one collector and run another"
);
}
if daemon_with_config(&name, false, serde_json::json!({})).is_some() {
anyhow::bail!(
"`{name}` is already a BUILT-IN collector of korp-collectors. Refusing to shadow \
it: an operator's roster would name one collector and run another"
);
}
external_registry()
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(name, Arc::new(factory));
Ok(())
}
/// Every externally-registered name, sorted. What `list` must show, so an operator can
/// see the collectors a foreign crate contributed alongside the built-in ones.
pub fn external_daemon_names() -> Vec<String> {
external_registry()
.lock()
.unwrap_or_else(|e| e.into_inner())
.keys()
.cloned()
.collect()
}
fn external_with_config(
name: &str,
enabled: bool,
config: serde_json::Value,
) -> Option<Arc<dyn Collector>> {
let f = external_registry()
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(name)
.cloned()?;
Some(f(enabled, config))
}
/// The daemon registry — every built-in collector daemon **plus every externally
/// registered one**, so korp iterates them to build the Settings sections + the infra
/// matrix + the startup daemon set (design §1). Honours the feature set (osm-cut under
/// `osm-daemon`). New collectors slot in here (build order §11 — the adopt-the-rest
/// phase); a collector from another crate arrives via
/// [`register_external_daemon`] instead.
pub fn daemon_registry() -> Vec<Arc<dyn Collector>> {
let mut collectors = builtin_daemon_registry();
// …and whatever a foreign crate contributed, in its default (disabled) shape. A
// registry that listed only the built-ins would make an externally registered
// collector runnable but invisible — declarable in a roster and absent from `list`.
for name in external_daemon_names() {
if let Some(c) = external_with_config(&name, false, serde_json::json!({})) {
collectors.push(c);
}
}
collectors
}
/// **Only the collectors compiled into THIS crate**, in their default (disabled) shape.
///
/// Split out from [`daemon_registry`] so that "how many collectors does this build
/// contain" stays a property of the build. It was one number until an external registry
/// existed, and a size assertion over the combined list would then depend on whatever
/// else the process had registered — which is how a feature-gating test starts failing
/// for reasons that have nothing to do with features.
#[allow(clippy::vec_init_then_push)] // conditional (feature-gated) pushes; not a fixed literal
pub fn builtin_daemon_registry() -> Vec<Arc<dyn Collector>> {
#[allow(unused_mut)]
let mut collectors: Vec<Arc<dyn Collector>> = Vec::new();
#[cfg(feature = "osm-daemon")]
collectors.push(Arc::new(crate::osm_cut::OsmCutDaemon::new()));
// WAVE 1 (design §11) — the first three existing collectors adopted to the daemon
// pattern, warehouse-sink (Iceberg) rather than S3-sink. Pure (`parsers`), so they
// register under `daemon` alone (no osm-katana). The remaining ~17 slot in here.
collectors.push(Arc::new(crate::iceberg_daemons::OsmAdminDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::DcatDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::PxWebDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::NordPoolDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::SdmxDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::UsgsDaemon::new()));
// WAVE 2, batch 2 — openaq / erddap / wfs adopted to the warehouse-sink pattern.
collectors.push(Arc::new(crate::iceberg_daemons::OpenAqDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::ErddapDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::WfsDaemon::new()));
// WAVE 2, batch 3 — energy-charts / rss-atom / the ENTSO-E trio (base / day-ahead /
// generation) adopted to the warehouse-sink pattern.
collectors.push(Arc::new(crate::iceberg_daemons::EnergyChartsDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::RssDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::EntsoeDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::EntsoeDayAheadDaemon::new()));
collectors.push(Arc::new(
crate::iceberg_daemons::EntsoeGenerationDaemon::new(),
));
// WAVE 2, batch 4 — GTFS-Realtime vehicles (pure protobuf reader, under `daemon`)
// and GTFS static stops (pure-Rust ZIP + csv, gated behind the `gtfs` feature just
// as osm-cut is gated behind `osm-daemon`).
collectors.push(Arc::new(crate::iceberg_daemons::GtfsRtVehiclesDaemon::new()));
#[cfg(feature = "gtfs")]
collectors.push(Arc::new(crate::iceberg_daemons::GtfsStopsDaemon::new()));
// WAVE 2, batch 5 — World Bank / GBIF / EEA-WISE open-data collectors (each rides
// the generic `rest::parse_records` JSON mapper, so all pure `parsers` under `daemon`).
collectors.push(Arc::new(crate::iceberg_daemons::WorldBankDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::GbifDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::EeaWiseDaemon::new()));
// WAVE 2, batch 6 — the Swedish real-time public-sector sources: polisen-events and
// smhi-weather (plain-GET JSON) and trafikverket-trainpos (a POST-body XML query,
// hand-written over the same reconcile). All pure `parsers` under `daemon`.
collectors.push(Arc::new(crate::iceberg_daemons::PolisenDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::SmhiWeatherDaemon::new()));
collectors.push(Arc::new(crate::iceberg_daemons::TrafikverketDaemon::new()));
collectors
}
/// Look up a registered daemon collector by name.
///
/// A name that a build has since RENAMED is migrated first ([`migrated_name`]), so a
/// row persisted under an old key still resolves to the collector that replaced it —
/// instead of silently becoming "not a registered daemon" and switching itself off.
pub fn daemon_by_name(name: &str) -> Option<Arc<dyn Collector>> {
let name = migrated_name(name);
daemon_registry().into_iter().find(|c| c.name() == name)
}
/// **Daemons renamed to match their registry twin** (`old`, `new`).
///
/// Every entry here exists because ONE source had two spellings: the short source
/// name on the daemon side and the qualified name on the collector side. The name is
/// the only thing tying a daemon to its `Collector` twin, so while they differed the
/// twin was unreachable and everything crossing that seam — freshness, read marks,
/// business keys, the graph mapping — skipped the pair without a word.
///
/// `worldbank` is the entry that shows why this had to be a rename and not a fuzzy
/// lookup: the short name is a prefix of TWO collectors (`worldbank-indicators`, the
/// REST indicators API the daemon actually fetches, and `worldbank-sdmx`, a different
/// API), so no lookup could have guessed which one was meant.
///
/// Landed 2026-09-03, measured by `tests/daemon_names.rs`.
pub const RENAMED_DAEMONS: &[(&str, &str)] = &[
("dcat", "dcat-ap-se"),
("pxweb", "pxweb-jsonstat2"),
("worldbank", "worldbank-indicators"),
("gbif", "gbif-occurrences"),
// The FIFTH, and the one my first census missed: it ran under `--features daemon`,
// where the twin is not compiled at all, so `osm-admin` read as a daemon with no
// twin rather than one whose twin it cannot name. A feature-gated collector is
// exactly where a name census goes blind, and this is what that blindness looks
// like. Found by a co-agent running the same test under `--all-features`.
("osm-admin", "osm-admin-boundaries"),
];
/// **The ONE rename table for persisted collector keys.**
///
/// A collector is a config key in someone's `config.json`, so renaming one is a
/// migration, not an edit. Both lookups here and korp's settings load route through
/// this function, so a row cannot resolve to one collector on one side and none on the
/// other. Every name with no rename is returned unchanged.
///
/// [`RENAMED_DAEMONS`] applies in EVERY build. It used to be that the whole function
/// was `osm_cut`'s single rename behind `#[cfg(feature = "osm-daemon")]`, which meant a
/// build without that feature migrated nothing at all — fine while osm-cut owned the
/// only rename, and a silent hole the moment a second one existed.
pub fn migrated_name(name: &str) -> &str {
// osm-cut's rename stays beside the collector that was renamed; it is only
// reachable in a build that HAS that collector to migrate to.
#[cfg(feature = "osm-daemon")]
let name = crate::osm_cut::migrate_collector_name(name);
RENAMED_DAEMONS
.iter()
.find(|(old, _)| *old == name)
.map(|(_, new)| *new)
.unwrap_or(name)
}
/// **Construct a registered daemon by name WITH an explicit `enabled` + `config`** —
/// what korp builds from the persisted settings before spawning [`Collector::run`]. In
/// contrast to [`daemon_by_name`] (which returns the disabled, default-config registry
/// entry), this overlays the operator's config so the metal spawn is generic over every
/// collector (the registry-driven spawn, design §11). `None` for an unknown name.
pub fn daemon_with_config(
name: &str,
enabled: bool,
config: serde_json::Value,
) -> Option<Arc<dyn Collector>> {
let name = migrated_name(name);
#[cfg(feature = "osm-daemon")]
if name == crate::osm_cut::NAME {
return Some(Arc::new(crate::osm_cut::OsmCutDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::OsmAdminDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::OsmAdminDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::DcatDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::DcatDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::PxWebDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::PxWebDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::NordPoolDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::NordPoolDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::SdmxDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::SdmxDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::UsgsDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::UsgsDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::OpenAqDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::OpenAqDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::ErddapDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::ErddapDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::WfsDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::WfsDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::EnergyChartsDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::EnergyChartsDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::RssDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::RssDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::EntsoeDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::EntsoeDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::EntsoeDayAheadDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::EntsoeDayAheadDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::EntsoeGenerationDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::EntsoeGenerationDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::GtfsRtVehiclesDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::GtfsRtVehiclesDaemon::with_config(enabled, config),
));
}
#[cfg(feature = "gtfs")]
if name == crate::iceberg_daemons::GtfsStopsDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::GtfsStopsDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::WorldBankDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::WorldBankDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::GbifDaemon::NAME {
return Some(Arc::new(crate::iceberg_daemons::GbifDaemon::with_config(
enabled, config,
)));
}
if name == crate::iceberg_daemons::EeaWiseDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::EeaWiseDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::PolisenDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::PolisenDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::SmhiWeatherDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::SmhiWeatherDaemon::with_config(enabled, config),
));
}
if name == crate::iceberg_daemons::TrafikverketDaemon::NAME {
return Some(Arc::new(
crate::iceberg_daemons::TrafikverketDaemon::with_config(enabled, config),
));
}
// The built-ins are asked FIRST, so a foreign registration can never shadow one even
// if the collision check in `register_external_daemon` were somehow bypassed. Two
// guards for one property, deliberately: this one is structural and cannot be
// argued with.
external_with_config(name, enabled, config)
}
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots_blocking(b: reqwest::blocking::ClientBuilder) -> reqwest::blocking::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn phase_labels_progress_and_health() {
assert_eq!(Phase::Offline.label(), "offline");
assert_eq!(Phase::Downloading { pct: 0.42 }.label(), "downloading 42%");
assert_eq!(
Phase::Cutting {
region: "sweden".into(),
pct: 0.5
}
.label(),
"cutting sweden 50%"
);
assert_eq!(Phase::Online.label(), "online");
assert_eq!(Phase::Downloading { pct: 0.42 }.progress(), Some(0.42));
assert_eq!(
Phase::Cutting {
region: "x".into(),
pct: 0.5
}
.progress(),
Some(0.5)
);
assert_eq!(Phase::Indexing.progress(), None);
assert_eq!(Phase::Online.progress(), None);
// Only Online is healthy; "not online yet" is a fact, not unhealthy-as-failed.
assert!(Phase::Online.healthy());
assert!(!Phase::Downloading { pct: 0.1 }.healthy());
assert!(!Phase::Failed { reason: "x".into() }.healthy());
assert!(Phase::Indexing.is_working());
assert!(!Phase::Online.is_working());
}
/// The matrix legend (design §8): green=Online, amber=working/not-yet,
/// grey=off (disabled), red=failed. A disabled collector is NEVER red.
#[test]
fn matrix_colour_maps_phase_and_enabled() {
assert_eq!(Phase::Online.matrix_color(true), MatrixColor::Green);
assert_eq!(
Phase::Downloading { pct: 0.1 }.matrix_color(true),
MatrixColor::Amber
);
assert_eq!(Phase::Indexing.matrix_color(true), MatrixColor::Amber);
assert_eq!(
Phase::Cutting {
region: "s".into(),
pct: 0.2
}
.matrix_color(true),
MatrixColor::Amber
);
assert_eq!(
Phase::Offline.matrix_color(true),
MatrixColor::Amber,
"enabled but offline = working toward online"
);
assert_eq!(
Phase::Failed { reason: "x".into() }.matrix_color(true),
MatrixColor::Red
);
// Disabled ⇒ grey regardless of phase — even a would-be failure is just off.
assert_eq!(Phase::Online.matrix_color(false), MatrixColor::Off);
assert_eq!(
Phase::Failed { reason: "x".into() }.matrix_color(false),
MatrixColor::Off
);
assert!(MatrixColor::Green.is_green());
assert!(!MatrixColor::Amber.is_green());
assert_eq!(MatrixColor::Red.rgb(), (220, 80, 80));
}
#[test]
fn phase_serde_round_trips_tagged() {
for p in [
Phase::Offline,
Phase::Downloading { pct: 0.3 },
Phase::Indexing,
Phase::Cutting {
region: "sweden".into(),
pct: 0.9,
},
Phase::Online,
Phase::Degraded {
reason: "stale".into(),
},
Phase::Failed {
reason: "bad planet".into(),
},
] {
let s = serde_json::to_string(&p).unwrap();
assert!(s.contains("\"state\""), "tagged wire shape: {s}");
let back: Phase = serde_json::from_str(&s).unwrap();
assert_eq!(back, p);
}
}
#[test]
fn mem_object_store_round_trips_and_lists_by_prefix() {
let s = MemObjectStore::new();
s.put("nordisk-maps", "sweden/tiles/z6/33/18.parquet", b"tileA")
.unwrap();
s.put("nordisk-maps", "sweden/tiles/z9/1/2.parquet", b"tileB")
.unwrap();
s.put(
"nordisk-maps",
"sweden/tiles/manifest.json",
br#"{"z":[6,9]}"#,
)
.unwrap();
s.put("other", "x", b"z").unwrap();
assert_eq!(
s.get("nordisk-maps", "sweden/tiles/z6/33/18.parquet")
.unwrap()
.as_deref(),
Some(&b"tileA"[..])
);
assert!(s.get("nordisk-maps", "nope").unwrap().is_none());
let tiles = s.list("nordisk-maps", "sweden/tiles/z").unwrap();
assert_eq!(
tiles.len(),
2,
"prefix list scopes to the z tiles, not the manifest"
);
assert_eq!(s.count("nordisk-maps"), 3);
assert_eq!(s.count("other"), 1);
// DELETE is the seam a stale-tile retire needs (it did not exist until
// 2026-08-02): the key is gone from both `get` and `list`, the sibling keys
// and the other bucket are untouched, and a second delete of the same key is
// a clean idempotent Ok (so a retire loop can be re-run).
s.delete("nordisk-maps", "sweden/tiles/z6/33/18.parquet")
.unwrap();
assert!(
s.get("nordisk-maps", "sweden/tiles/z6/33/18.parquet")
.unwrap()
.is_none()
);
assert_eq!(
s.list("nordisk-maps", "sweden/tiles/z").unwrap(),
vec!["sweden/tiles/z9/1/2.parquet".to_string()]
);
assert_eq!(s.count("nordisk-maps"), 2, "only the deleted key went");
assert_eq!(s.count("other"), 1, "the other bucket is untouched");
s.delete("nordisk-maps", "sweden/tiles/z6/33/18.parquet")
.unwrap();
s.delete("nordisk-maps", "never-existed").unwrap();
assert_eq!(
s.count("nordisk-maps"),
2,
"deleting an absent key changes nothing"
);
}
/// The BIOS bus (design §6): a collector "tunes in over time" — a "not online
/// yet" fact, then ONLINE. The aggregator keeps append order, flips `is_online`
/// only on the ONLINE fact, and renders the POST-style banner.
#[test]
fn bios_bus_reports_not_online_yet_then_online_in_order() {
let log = BiosLog::new();
let bus = log.bus();
assert!(!log.is_online("map"), "unknown collector is not online");
bus.not_online_yet("map", "downloading 12%");
bus.not_online_yet("map", "indexing regions");
assert!(
!log.is_online("map"),
"'not online yet' never flips online true"
);
bus.online("energy");
bus.online("map");
assert!(log.is_online("map"));
assert!(log.is_online("energy"));
let facts = log.snapshot();
assert_eq!(facts.len(), 4);
// Append order preserved via the monotonic seq.
assert!(
facts.windows(2).all(|w| w[0].seq < w[1].seq),
"append order is monotonic"
);
assert_eq!(facts[0].message, "not online yet (downloading 12%)");
assert!(facts[3].online && facts[3].collector == "map");
let banner = log.render();
assert!(
banner.contains("collector map: not online yet (downloading 12%)"),
"{banner}"
);
assert!(banner.contains("collector map: ONLINE"), "{banner}");
}
#[test]
fn shutdown_flag_is_cooperative_and_shared() {
let s = Shutdown::new();
let c = s.clone();
assert!(!s.is_stopped());
c.stop();
assert!(
s.is_stopped(),
"clones share one flag (a config toggle stops the daemon)"
);
}
// ── A mock daemon exercising the whole trait trio + ctx round-trip ──────────
struct MockDaemon {
enabled: bool,
phase: Mutex<Phase>,
}
impl CollectorConfig for MockDaemon {
fn name(&self) -> &str {
"mock"
}
fn enabled(&self) -> bool {
self.enabled
}
fn schema(&self) -> ConfigSchema {
ConfigSchema::new(vec![
ConfigField::of(
"regions",
"Regions",
"pick",
FieldKind::RegionPickList {
catalog: vec!["sweden".into()],
},
),
ConfigField::of("zooms", "Zooms", "6,9,12", FieldKind::Zooms),
])
}
}
impl CollectorState for MockDaemon {
fn phase(&self) -> Phase {
self.phase.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
fn facet(&self) -> serde_json::Value {
serde_json::json!({ "collector": "mock", "phase": self.phase().label() })
}
}
#[async_trait::async_trait]
impl Collector for MockDaemon {
async fn run(self: Arc<Self>, ctx: CollectorCtx) {
// A tiny state machine: Offline → Downloading → Cutting → Online, then
// hold Online until asked to stop.
for p in [
Phase::Offline,
Phase::Downloading { pct: 0.5 },
Phase::Cutting {
region: "sweden".into(),
pct: 0.5,
},
Phase::Online,
] {
*self.phase.lock().unwrap_or_else(|e| e.into_inner()) = p.clone();
ctx.set_phase(&p);
ctx.log.line(p.label());
if p.is_online() {
// Prove the bucket seam works: write one tile.
ctx.bucket
.put("nordisk-maps", "sweden/tiles/z6/0/0.parquet", b"t")
.unwrap();
}
}
while !ctx.shutdown.is_stopped() {
tokio::task::yield_now().await;
break; // test: don't spin forever
}
}
}
/// **The daemon trait trio + ctx drive a mock collector to ONLINE.** The phase
/// sink records the full transition sequence, the BIOS bus tunes in to ONLINE,
/// the matrix goes green, and a tile lands in the (in-memory) bucket. This is the
/// object-safe `Vec<Arc<dyn Collector>>` path korp uses.
#[tokio::test]
async fn mock_daemon_reaches_online_through_the_ctx_seams() {
let (phase_sink, phases) = PhaseSink::recorder();
let bios = BiosLog::new();
let bucket = Arc::new(MemObjectStore::new());
let ctx = CollectorCtx {
name: "mock".into(),
disk_dir: std::env::temp_dir(),
config: serde_json::json!({}),
bucket: bucket.clone(),
warehouse: Arc::new(MemWarehouse::new()),
fetch: Fetcher::noop(),
phase: phase_sink,
log: LogSink::noop(),
bios: bios.bus(),
shutdown: Shutdown::new(),
secret: std::sync::Arc::new(crate::daemon::EnvSecretStore::new()),
marks: std::sync::Arc::new(crate::freshness::MemMarkStore::new()),
nudge: Nudge::new(),
jobs: <dyn ChildJobs>::noop(),
};
let daemon: Arc<dyn Collector> = Arc::new(MockDaemon {
enabled: true,
phase: Mutex::new(Phase::Offline),
});
daemon.clone().run(ctx).await;
// Reached Online, and the matrix cell is green.
assert_eq!(daemon.phase(), Phase::Online);
assert!(daemon.healthy());
assert_eq!(matrix_color(&*daemon), MatrixColor::Green);
assert_eq!(daemon.facet()["phase"], "online");
// The phase sink saw the whole state-machine sequence, ending Online.
let seq = phases.lock().unwrap().clone();
assert_eq!(seq.first(), Some(&Phase::Offline));
assert_eq!(seq.last(), Some(&Phase::Online));
// The BIOS bus tuned in: "not online yet" facts, then ONLINE.
assert!(
bios.is_online("mock"),
"the daemon tuned in to ONLINE on the bus"
);
assert!(bios.render().contains("collector mock: ONLINE"));
// A tile landed in the bucket via the ObjectStore seam.
assert_eq!(bucket.count("nordisk-maps"), 1);
}
#[test]
fn registry_is_feature_gated() {
// The warehouse-sink daemons register under `daemon` alone (pure parsers, no
// osm-katana): the three WAVE-1 daemons (osm-admin/dcat/pxweb), the WAVE-2
// batch-1 adoptions (nordpool/eurostat-sdmx/usgs-earthquakes), the batch-2
// adoptions (openaq-air-quality/erddap-emodnet-physics/wfs), the batch-3
// adoptions (energy-charts-power/rss-atom/entsoe/entsoe_day_ahead/
// entsoe_generation) and the batch-4 GTFS-Realtime vehicles (a pure protobuf
// reader). gtfs-stops adds one more under the `gtfs` feature (ZIP + csv);
// osm-cut adds one more under `osm-daemon`.
let daemon_names = [
"osm-admin-boundaries",
"dcat-ap-se",
"pxweb-jsonstat2",
"nordpool",
"eurostat-sdmx",
"usgs-earthquakes",
"openaq-air-quality",
"erddap-emodnet-physics",
"wfs",
"energy-charts-power",
"rss-atom",
"entsoe",
"entsoe_day_ahead",
"entsoe_generation",
"gtfs-rt-vehicles",
"worldbank-indicators",
"gbif-occurrences",
"eea-wise-water",
"polisen-events",
"smhi-weather",
"trafikverket-trainpos",
];
// BUILT-INS only: the combined `daemon_registry()` also carries whatever another
// crate registered in this process, and a build-shape assertion must not depend
// on that.
let reg = builtin_daemon_registry();
for name in daemon_names {
assert!(
reg.iter().any(|c| c.name() == name),
"{name} self-registers under daemon"
);
assert!(daemon_by_name(name).is_some(), "{name} resolvable by name");
}
// Base count under `daemon` alone: the 21 pure-parser warehouse-sink daemons
// (the 15 earlier waves + the batch-5 worldbank / gbif / eea-wise-water trio +
// the batch-6 polisen-events / smhi-weather / trafikverket-trainpos trio). The
// `gtfs` and `osm-daemon` features each add exactly one more (gtfs-stops /
// osm-cut), so the expected registry size is feature-additive.
let mut expected = 21;
#[cfg(feature = "gtfs")]
{
expected += 1;
assert!(
reg.iter().any(|c| c.name() == "gtfs-stops"),
"gtfs-stops self-registers under gtfs"
);
assert!(daemon_by_name("gtfs-stops").is_some());
}
#[cfg(not(feature = "gtfs"))]
assert!(
daemon_by_name("gtfs-stops").is_none(),
"gtfs-stops absent without the gtfs feature"
);
#[cfg(feature = "osm-daemon")]
{
expected += 1;
assert!(
reg.iter().any(|c| c.name() == "osm-cut"),
"osm-cut self-registers under osm-daemon"
);
assert!(daemon_by_name("osm-cut").is_some());
// …and the name it was persisted under before the rename still resolves,
// through the ONE migration table. A live server's config row must not
// silently become "not a registered daemon".
assert_eq!(
daemon_by_name("osm-tiles").map(|c| c.name().to_string()),
Some("osm-cut".to_string()),
"the legacy key migrates rather than vanishing"
);
}
#[cfg(not(feature = "osm-daemon"))]
{
assert!(
daemon_by_name("osm-cut").is_none(),
"osm-cut absent without the osm-daemon feature"
);
assert!(
daemon_by_name("osm-tiles").is_none(),
"…and so is the name it used to have"
);
}
assert_eq!(
reg.len(),
expected,
"registry size is feature-additive (21 base + gtfs + osm-daemon)"
);
// daemon_with_config overlays enabled + config generically (the metal spawn seam).
for name in daemon_names {
let d = daemon_with_config(name, true, serde_json::json!({ "table": "t" }))
.unwrap_or_else(|| panic!("{name} constructs with config"));
assert!(d.enabled(), "{name}: enabled overlaid");
assert_eq!(d.name(), name);
}
#[cfg(feature = "gtfs")]
{
let d = daemon_with_config("gtfs-stops", true, serde_json::json!({ "table": "t" }))
.expect("gtfs-stops constructs with config");
assert!(d.enabled());
assert_eq!(d.name(), "gtfs-stops");
}
assert!(daemon_with_config("nope", true, serde_json::Value::Null).is_none());
}
/// ★ **RED PROBE — a `Fetcher::http()` must be constructible AND droppable INSIDE an
/// async context.**
///
/// `reqwest::blocking::Client` owns a tokio runtime, and creating or dropping one while
/// another runtime is current panics with *"Cannot drop a runtime in a context where
/// blocking is not allowed"*. That is not hypothetical: a supervisor that set
/// `ctx.fetch = Fetcher::http()` inside an async block and handed the ctx to `run` killed
/// the collector on the DROP — before a single byte was fetched — and the failure looked
/// like the collector, not like the seam.
///
/// The probe fails if the shared-client discipline is ever reverted, which is the point:
/// nothing else in the suite builds a fetcher from inside a runtime.
#[test]
fn a_fetcher_is_safe_to_build_and_drop_inside_a_tokio_context() {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()
.unwrap();
rt.block_on(async {
let f = Fetcher::http();
// A second one, so the OnceLock's initialised path is exercised too.
let g = Fetcher::http();
drop(f);
drop(g);
});
// And the same across a spawned task, which is where the real supervisor lives.
rt.block_on(async {
tokio::spawn(async {
let mut ctx =
CollectorCtx::in_memory("probe", std::env::temp_dir(), serde_json::json!({}));
ctx.fetch = Fetcher::http();
drop(ctx);
})
.await
.expect("building and dropping a fetcher inside a spawned task must not panic");
});
}
// ── The EXTERNAL registry ────────────────────────────────────────────────
/// A minimal foreign collector, standing in for one that lives in another crate
/// entirely (moln's `moln-payload` is the first real one).
struct ForeignCollector {
name: String,
enabled: bool,
config: serde_json::Value,
}
impl CollectorConfig for ForeignCollector {
fn name(&self) -> &str {
&self.name
}
fn enabled(&self) -> bool {
self.enabled
}
fn schema(&self) -> ConfigSchema {
ConfigSchema::new(vec![])
}
}
impl CollectorState for ForeignCollector {
fn phase(&self) -> Phase {
Phase::Offline
}
fn facet(&self) -> serde_json::Value {
self.config.clone()
}
}
#[async_trait::async_trait]
impl Collector for ForeignCollector {
async fn run(self: Arc<Self>, _ctx: CollectorCtx) {}
}
fn foreign(name: &'static str) -> impl Fn(bool, serde_json::Value) -> Arc<dyn Collector> {
move |enabled, config| {
Arc::new(ForeignCollector {
name: name.to_string(),
enabled,
config,
})
}
}
/// ★ **A collector from another crate can be declared, resolved and listed.**
///
/// This is the door `nursery.rs:6-8` promised and `daemon_with_config`'s closed
/// if-chain did not provide: before this hook, a foreign name's ONLY outcome was the
/// "no collector named `x`" refusal.
///
/// The assertion is on APPLIED OUTPUT — the resolved collector's own `enabled` and
/// `facet()` carry the values the caller passed, so a factory that ignored them
/// cannot pass.
#[test]
fn a_collector_from_another_crate_resolves_through_the_registry() {
const N: &str = "test-external-resolves";
assert!(
daemon_with_config(N, true, serde_json::json!({})).is_none(),
"the name must be unknown before registration, or this proves nothing"
);
register_external_daemon(N, foreign(N)).expect("register a foreign collector");
let cfg = serde_json::json!({"pool_dir": "/srv/moln/pool"});
let c = daemon_with_config(N, true, cfg.clone())
.expect("a registered foreign collector must resolve");
assert_eq!(c.name(), N);
assert!(c.enabled(), "the `enabled` flag reached the factory");
assert_eq!(c.facet(), cfg, "and so did the config");
assert!(external_daemon_names().iter().any(|n| n == N));
assert!(
daemon_registry().iter().any(|c| c.name() == N),
"an externally registered collector must also be LISTED — runnable but \
invisible is worse than absent"
);
// Listed in its DEFAULT shape, not the one the last resolve asked for.
let listed = daemon_registry()
.into_iter()
.find(|c| c.name() == N)
.unwrap();
assert!(
!listed.enabled(),
"the registry entry is off until declared on"
);
}
/// ★ **RED — a foreign registration cannot shadow a built-in.** Last-write-wins here
/// would let another crate take the name `dcat`, and an operator's roster would name
/// one collector while running another.
#[test]
fn a_foreign_registration_cannot_shadow_a_builtin() {
let e =
register_external_daemon(crate::iceberg_daemons::DcatDaemon::NAME, foreign("shadow"))
.expect_err("shadowing a built-in must refuse");
let m = e.to_string();
assert!(m.contains("BUILT-IN"), "{m}");
assert!(m.contains("run another"), "{m}");
// …and the built-in still resolves to itself, not to the impostor.
let c = daemon_with_config(
crate::iceberg_daemons::DcatDaemon::NAME,
true,
serde_json::json!({}),
)
.unwrap();
assert_eq!(c.name(), crate::iceberg_daemons::DcatDaemon::NAME);
assert_ne!(
c.facet(),
serde_json::json!({}),
"the real dcat daemon has a facet of its own; the stub's is the bare config"
);
}
/// ★ **RED — registering the same foreign name twice is REFUSED**, with a message
/// that points at the other crate rather than at this one. An empty name is refused
/// too: a collector nobody can name is a collector nobody can declare.
#[test]
fn a_duplicate_or_nameless_foreign_registration_is_refused() {
const N: &str = "test-external-duplicate";
register_external_daemon(N, foreign(N)).expect("first registration");
let e = register_external_daemon(N, foreign(N))
.expect_err("a second registration of the same name must refuse");
assert!(
e.to_string().contains("another crate in this process"),
"{e}"
);
let e = register_external_daemon(" ", foreign("blank"))
.expect_err("an empty name must refuse");
assert!(e.to_string().contains("must have a name"), "{e}");
}
}