<!-- nornir:class: design -->
# korp Collectors — the daemon architecture (design)
> **Collector vs Pipeline.** A **collector deals with OUTSIDE data** — it ingests an
> external source into korp. A korp **pipeline** is **in-house data movement +
> transformation** (SDP, Iceberg→FalkorDB). Collectors = ingress daemons; pipelines =
> internal transforms. This repo is collectors only.
A **collector** is a **continuous daemon** that *strives* to reach and hold **ONLINE**
for everything in its config, self-healing across restarts. It is a **state machine**,
not a one-shot job. `osm-cut` (planet → regions → the lettered a–g cut) is the first implementor; the
existing Iceberg collectors (DCAT-AP-SE, PxWeb, ENTSO-E, REST) are **redone** to this
design.
## 1. The traits (object-safe, so new collectors just slot in)
```
trait CollectorConfig // what you SET → Settings tab
fn enabled(&self) -> bool
fn schema() -> ConfigSchema // fields, incl. multi-target (region pick-list)
(serde in/out of korp's settings JSON: collectors.<name> = { enabled, …custom })
trait CollectorState // what it's DOING → infra status matrix + click-expand
fn phase(&self) -> Phase // the state-machine value (see §3)
fn healthy(&self) -> bool // green when Online+healthy
fn facet(&self) -> &dyn Facet // its OWN custom UI area (state_json + ui)
trait Collector : CollectorConfig + CollectorState // the daemon
fn run(&mut self, ctx: CollectorCtx) -> ! // the never-ending loop
// ticks its state machine toward ONLINE for every configured target
```
`CollectorCtx` hands the collector its infra: **both** storage backends (§4), a **jera
daemon-job handle** (§5), the **normal log**, and the **BIOS bus** (§6).
## 2. Continuous daemon (not a one-shot)
The collector loop runs forever: reconcile config → do the next needed work → sleep/watch
→ repeat. For `osm-cut`: keep the planet fresh, keep every configured region cut into
`<work_dir>/<region>/`, re-cut when a region is added or the planet updates. "Idle & Online" is just the
steady state of the same loop.
## 3. State machine (pure facts — BIOS-style, never an "error" to be not-yet-online)
Per target **and** overall:
```
Offline → Downloading(planet, %) → Indexing(planet → region catalog)
→ Cutting(region, "<letter> <step>") → Online (+ Degraded/Failed{reason} off any)
```
Being `Downloading` / `not online yet` is a **normal fact**, not a failure. Transitions
are appended to the BIOS bus (§6). The machine is **restart-safe**: on korp-server
restart it resumes from what's already on disk / in the bucket (batched files are reused,
not re-fetched).
## 4. Storage — every collector gets BOTH
Some data is servable (belongs in S3); some needs random access and is dumb to keep in S3
(the planet file, indexes). So `CollectorCtx` provides:
- **S3 bucket(s)** in the rustfs infra — servable artifacts. `osm-cut` uses none: its
output is FILES under `<work_dir>/<region>/`, which is what korp-server reads.
- **A per-collector on-disk dir** under korp's user dir — `~/.korp/collectors/<name>/` —
the planet `.pbf`, the region catalog, random-access caches, batched intermediates.
## 5. jera — a TRUE never-ending daemon job (jera gets extended)
Today jera models fire-and-complete jobs. Add a first-class **daemon** job kind:
- a job **expected to run indefinitely** — a long `Running` with a **phase** + **progress**
+ a **healthy** flag; **never counts as failed for not completing**;
- may cycle sub-phases (`Downloading → Online → Watching → …`);
- **restart-safe** — survives korp-server restarts, resumes its phase;
- stoppable/startable from the config toggle.
One daemon job **per enabled collector**. Its phase/progress is exactly what the State
trait surfaces to the matrix.
## 6. Two logs — the work log AND the BIOS bus
- **Normal log** (the jera job log): detailed, per-collector — "planet 42%", "cut sweden
→ step e networks produced, 1.2 GB", warnings, errors. What you read to debug one collector.
- **BIOS status bus**: a coarse, append-only, **pure-fact** stream the *server* aggregates
at startup — like a hardware BIOS POST that **tunes in over time without blocking or
invalidating startup**:
```
collector map: not online yet (downloading planet 12%)
collector map: indexing regions…
collector energy: ONLINE
collector map: ONLINE
```
Startup does **not** wait for collectors; the bus just reports facts as each tunes in.
The infra matrix (§8) and a startup banner render this bus.
## 7. Region pre-list, sucked out of the planet (map collector)
After the torrent completes, **osm-katana enumerates the planet's regions/sections** (it's
fast, gatling-parallel) into a **cached region catalog** on disk. Then:
- Settings shows a **multi-select pick-list** of available regions;
- the collector answers **"what regions exist?"** from the cached catalog — **never
re-scanning the planet per query**;
- before the planet exists the catalog is empty → Settings shows "planet not downloaded
yet" (a fact, not an error). This is why the machine has an explicit `Indexing` phase.
## 8. Config (Settings) + Status (infra matrix) — the two faces
- **Settings ▸ Collectors**: per-collector **on/off** + custom config (region pick-list,
the last step to run, the clip, cache caps). Driven by the **Config trait**.
- **Infra ▸ Collectors matrix**: one cell per collector — **green = Online+healthy**,
**amber = working** (downloading/cutting), **grey = off**, **red = failed**. Each cell
**click-expands** the collector's **State-trait Facet** (its own UI: phase, progress,
the step letter it is on and a report per letter, last run). Driven by the **State
trait** + the BIOS bus.
## 9. Torrent code moves INTO the collector
The librqbit planet-torrent fetch (currently `korp/src/download.rs`) **moves into the
`osm-cut` collector** — the collector owns its own acquisition. `download.rs` becomes a
thin re-export or is absorbed.
## 10. Bencher — prove the tiling uses ALL cores via gatling (NOT rayon)
Add a bencher (osm-katana), **heavy + non-heavy**, asserting the region cut (`convert`)
saturates all cores through **znippy-zoomies `gatling`** fork-join — the whole point of
osm-katana being rayon-free. Non-heavy = a small fixture (CI); heavy = a Sweden-scale cut.
Follows the bencher law (host = a new row, same-window comparisons only; "has a bencher"
is the gate, "fast enough" is not).
## 11. Adopt the existing collectors
DCAT-AP-SE / PxWeb / ENTSO-E / REST are **redone** as `Collector` daemons (Config + State
+ run-loop), sink to their Iceberg tables as before, and appear in the same matrix. The
pattern is generic; `osm-cut` (files on disk) and the Iceberg collectors (warehouse-sink)
differ only in `CollectorCtx` usage.
**WAVE 1 — DONE** (`src/iceberg_daemons.rs`, feature `daemon`): `osm-admin`, `dcat`,
`pxweb` adopted as the first three warehouse-sink daemons; `daemon_registry()` now lists
**4** (osm-cut + these three, under `osm-daemon`; the three alone under `daemon`). Two
new `CollectorCtx` seams make the pattern generic over both sink kinds:
- **`WarehouseSink`** — the Iceberg counterpart of `ObjectStore`: `append(table, batch)`.
Since 2026-09-01 it is ALSO the backing of the one-shot path: `BatchSink::warehouse(sink,
table)` (feature `daemon`) hands every `flush`ed batch to the host's `append` on the
blocking pool and reports the host's count as `landed()`; `run_with_sink(collector,
cfg, sink)` is the runner a host calls so EVERY `Collector` (osm, csv, fixedwidth,
mainframe, trafikverket, gribnc, the REST presets) lands through the host's seam —
in korp, `Streaming::land` on the configured backend row (`collect::RowSink`). `run(dir)`
+ `BatchSink::iceberg` (the leaf opening skade itself) stay for standalone consumers
and the fixture tests. `znippy_pack` emits no `RecordBatch` (archive bytes) and does not
go through it; `osm_cut` emits no MAP `RecordBatch` either — it writes FILES — but it
does land one SUMMARY row per (region, step) into `osm_cut_steps` through this seam,
so the cut shows up in the 🏬 Warehouse like every other collector. Gate:
`sink::warehouse_backing_tests` (a swallowing sink reports `landed() == 0`, never the
push tally).
korp backs it with a skade `Table` (`BatchSink::iceberg` → `fast_append`); tests with
`MemWarehouse`. This is the "sink to its Iceberg table via a warehouse seam" the adopt
step needed (S3-sink daemons ignore it; warehouse-sink daemons ignore `bucket`).
- **`Fetcher`** — 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>` carrying its authentication key) rides the same seam as
the plain-GET sources. The run-loop is proven headless with **no live gov API**: a test
wires `Fetcher::from_map(SYNTHETIC_*)` — which serves BOTH verbs by URL, so a POST-body
daemon is exercised offline against the same fixture — and korp wires the rustls HTTP
core. A daemon selects its verb through the local `Source` enum (`Get` / `Post`); the one
`reconcile_iceberg`/`daemon_run_loop` body handles both, so a POST source is not a second
skeleton — only its request shape differs.
- **`SecretStore`** — where a daemon's **secret** (API key, token, password) comes from, so
NO daemon calls `std::env::var` (or reads a config literal) directly and a vault can be
swapped in later WITHOUT the leaf gaining a vault dependency. Object-safe:
`fn secret(&self, name: &str) -> Option<SecretString>`, injected on `CollectorCtx.secret`.
The value is a **`SecretString`** — a wrapper whose `Debug`/`Display` render a fixed
`***redacted***`, whose bytes are zeroized on drop, and whose value is readable only at
the moment of use via `expose()` (so a stray `{:?}` / log line / panic cannot leak it; the
leaf needs no `zeroize` crate for this). The leaf ships **`EnvSecretStore`** (an inline
literal checked first, then the process env — today's behavior, unchanged, and the default
`CollectorCtx::in_memory` wires). `None` for a needed secret is a loud `Failed`, never a
silent green: the daemon re-resolves EACH tick (a key absent now may be leased later) and
never fetches without it. **`trafikverket-trainpos`** is the first consumer — its key
resolves through the store (an explicit config `key` override wins, otherwise
`store.secret(key_env)`), and the resolved key is exposed only into the POST body, never
logged. *TODO(vault): the future host impl is a `VaultSecretStore` over **nornir-vault /
OpenBao** (`vault.g11.se`, per-site zone) reading `secret/collectors/<name>/<key>` — it
lives in korp (the HOST), injected via the ctx; this crate keeps only the trait + the env
default. See the `// TODO(vault):` at `SecretStore` in `src/daemon.rs`.*
Each adopted daemon reuses its existing pure parser (`dcat::parse_dcat_rdfxml`,
`pxweb::parse_jsonstat2`, `osm_admin::extract_admin_units`, `rest::parse_records`, …) and
reaches ONLINE through the shared `Offline → Downloading → Indexing → Online` skeleton
(empty-is-RED → `Failed`).
**WAVE 2 — DONE** (`src/iceberg_daemons.rs`, feature `daemon`): the mechanical follow-on,
in batches, each a new `iceberg_daemon!(name, url, table, parse, schema)` line —
**batch 1**: sdmx · usgs · nordpool; **batch 2**: openaq · erddap · wfs; **batch 3**:
energy-charts · rss · entsoe{,-day-ahead,-generation}; **batch 4**: gtfs-rt-vehicles ·
gtfs-stops; **batch 5**: worldbank · gbif · eea-wise-water; **batch 6**: the Swedish
real-time public-sector trio — **`polisen-events`** and **`smhi-weather`** (plain-GET JSON
through `rest::parse_records` + the matching `RestCollector` preset field map) and
**`trafikverket-trainpos`** (the one POST-body daemon: it builds the XML `<REQUEST>` with
its key from config `key` / env `key_env` and POSTs it, then parses
`/RESPONSE/RESULT/0/TrainPosition` — hand-written over the SAME shared reconcile, only its
`Source` is a `Post`). `daemon_registry()` now lists **21** under `daemon` (+ `gtfs-stops`
under `gtfs`, + `osm-cut` under `osm-daemon`). Every registered daemon self-registers into
`collector-register.redb` (`register.rs`), so a `RegisterReader` in another process (korp's
Collectors panel, or `korp-collectors status`) sees each one's phase + facet — proven
RED→GREEN per daemon in `tests/collector_daemons_register.rs`. The remaining follow-on
(gribnc · rest · mainframe presets) is the same `iceberg_daemon!` line.
---
### Build order
1. **jera daemon-job kind** (§5) — foundation.
2. **Collector traits + CollectorCtx + registry** (§1, §4, §6).
3. **`osm-cut`** as the first daemon (§2, §3, §7, §9) + its **bencher** (§10).
4. **Settings + infra matrix** (§8).
5. **Adopt the existing collectors** (§11).
(HISTORICAL, 2026-08: the urgent map **un-blank** landed FIRST and independently. The
whole tiled route it fixed has since been deleted — korp map v2, `e66f846..1d53b1b` — and
the map is served from the packed cut this collector produces.)
## §14 — the REQUIREMENT declaration (lane CP, 2026-09-02)
A daemon **declares what it needs before it can reach ONLINE**, through the same
`crate::requires::Requirements` the one-shot ingest `Collector` trait answers with:
```rust
fn requirements(&self) -> crate::requires::Requirements; // CollectorConfig, defaulted
```
The default reads the source out of the daemon's own `defaults()` (`url` / `endpoint` /
`source` / `base_url`) and declares no secret. **A daemon that reads
`CollectorCtx::secret` MUST override it and name the variable** — a daemon whose key is
absent can otherwise only report `Failed` at the wire, which is the state this declaration
exists to replace with `NeedsConfig`.
`SecretString` / `SecretStore` / `EnvSecretStore` MOVED to `crate::requires` (unchanged;
`daemon::` re-exports them, so every existing path still resolves). The seam had to be
readable without the `daemon` feature, because readiness answers for the ingest registry
too and a second copy behind a second gate would be two answers to "is this key set?".
`requires::SecretPresence` is the one-method question readiness asks; the blanket
`impl<T: SecretStore + ?Sized> SecretPresence for T` is the ONLY writer of the answer, so
a host that swaps in a vault store changes what the UI calls runnable without readiness
learning what a vault is.