nordisk/korp-collectors

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 flushed 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:

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.