nordisk/korp-collectors
src/osm_cut.rs
//! **`osm-cut` — the region-cut collector daemon** (feature `osm-daemon`).
//!
//! The map-data daemon (design `.nornir/collectors/osm-cut.md`): keep every configured
//! region CUT from the OSM planet into the packed source korp's map actually serves,
//! striving to hold **ONLINE**. It is the first implementor of the [`crate::daemon`]
//! architecture — a continuous state machine, not a one-shot job.
//!
//! # It used to build a tile pyramid. That was legacy.
//!
//! The owner, 2026-09-02: *"the collector osm tiles is legacy? should it not be the new 3
//! files from osm download?"* — yes. korp's map v2 **deleted** the pyramid
//! (`e66f846..1d53b1b`: the tiling gate, the tiled/monolith routes, `src/map_tiles.rs`
//! at 4 980 lines, the S3 tile store and 111.8 GB on disk); `korp/tests/no_tiles_law.rs`
//! forbids the word in a user-visible string, and the client speaks CELLS. This collector
//! kept cutting a `z6/z9/z12` pyramid into a bucket **nothing reads** — ~300 references
//! to a dead artefact.
//!
//! So the collector now produces what the map is served from: the lettered chain
//! `korp::pipeline` publishes, one letter per step, ending in the packed base table.
//!
//! ```text
//! Offline → Downloading(planet %) → Indexing(planet → region catalog)
//! → Cutting(region, step a..g) → Online
//! ```
//!
//! | step | what it produces |
//! |---|---|
//! | a `planet` | a complete planet source on disk |
//! | b `coords` | convert pass 1 — `nodes.parquet` + `node_coords.arrow` |
//! | c `tables` | convert pass 2 — `ways.parquet` + `relations.parquet` |
//! | d `relations` | relation clip + geometry assembly — `relations.resolved.parquet` |
//! | e `networks` | `networks/` — roads · rails · trails · rail_nodes · graph_edges · graph_nodes |
//! | f `packed` | the packed base table (native Geometry + `covering.bbox` + rungs) |
//! | g `address` | the address index — `address.parquet` |
//!
//! * **Acquisition lives here** (design §9): the planet is resolved from an already-recorded
//! file under the collector's own directory, then from a seed path the operator
//! **DECLARED** (`config.seed` / `config.seeds`, linked in through
//! [`crate::acquire::LocalSeed`]), and — under `osm-daemon-download` — torrented via
//! `osm-katana`'s librqbit fixture. There is no `$HOME` search: a declared path is
//! reviewable, and a fallback that always wins hides the transport it is a fallback for.
//! * **Region catalog** (design §7): after the planet exists, `osm-katana` enumerates
//! its admin-region NAMES (`stream_admin_region_names` — a bounded-memory streaming
//! relation scan, gatling-parallel, NO in-RAM NodeStore) into a cached
//! `catalog.json` on disk → the Settings region pick-list. Never re-scans per query —
//! but the cache is KEYED on the planet that produced it, so a planet swap re-scans
//! once rather than serving the old planet's roster for ever.
//! * **Steps a–c are osm-katana's** `convert`, driven here with the region's clip.
//! * **Steps d–g have exactly ONE implementation and it lives in korp** — the same
//! `regions::start_or_resume` chain korp-server runs at boot. This crate is korp-free
//! by charter, so korp injects it through the [`CutDriver`] seam (LAW 5: one writer,
//! never a second copy of the chain). A build with no driver reports d–g as
//! [`StepState::Unavailable`] **by name** rather than pretending the cut is whole.
//! * **It emits NO `RecordBatch` of map data** — it writes FILES under
//! `~/.korp/work/<region>/`. What it *does* land, through the ordinary warehouse
//! seam, is one **summary row per (region, step)**: region · step · rows · bytes ·
//! wall_ms · state. So the cut shows up in the 🏬 Warehouse like every other
//! collector, and the row says what it is.
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use anyhow::{Context, Result};
use crate::daemon::{
Collector, CollectorConfig, CollectorCtx, CollectorState, ConfigField, ConfigSchema, FieldKind,
Phase,
};
use crate::osm_cut_verify::DigestMode;
use crate::state;
/// The collector name (the `collectors.<name>` settings key + matrix cell).
pub const NAME: &str = "osm-cut";
/// **The plan/graph name the cut's adjacency lands under** — ONE name, no hash.
/// The viewport slice's `sweden_roads_<8 hex>` naming belongs to a BOUNDED cut and
/// must not leak into the pipeline's own graph.
pub const GRAPH_PLAN: &str = "sweden_roads";
/// **The name this collector was persisted under before it stopped building a pyramid.**
///
/// A live server's `~/.korp/config.json` carries `collectors."osm-tiles"` with the
/// operator's `enabled` flag and their region list in it. Renaming the collector without
/// carrying that row across would silently switch the map cut OFF on the next start —
/// the loudest possible regression from the quietest possible change. See
/// [`migrate_collector_name`].
pub const LEGACY_NAME: &str = "osm-tiles";
/// **Rename a persisted collector key to what this build calls it.** The ONE writer both
/// repos migrate through: the leaf's registry lookup ([`crate::daemon::daemon_with_config`])
/// and korp's settings load call this, so a config row cannot resolve to one collector on
/// one side and none on the other. Every other name is returned unchanged.
pub fn migrate_collector_name(name: &str) -> &str {
if name == LEGACY_NAME { NAME } else { name }
}
/// **The table the cut lands its per-step SUMMARY rows in.**
///
/// The cut emits no map `RecordBatch` — it writes files. But a producer that lands
/// nothing is invisible in the 🏬 Warehouse, so each finished step lands exactly one row
/// (region · step · state · rows · bytes · wall_ms · at_unix) through the ordinary
/// [`crate::daemon::WarehouseSink`] seam. That is a receipt for work done, not a
/// re-publication of the map.
pub const SUMMARY_TABLE: &str = "osm_cut_steps";
/// The file name a linked-in seed is given inside the collector's own directory, when a
/// declared seed path has none of its own. **Not a search path**: a seed is acquired only
/// from a path the operator declared in `config.seed` / `config.seeds`.
pub const DEFAULT_SEED: &str = "planet-260427.osm.bz2";
/// The smallest thing that can be a planet. A declared seed shorter than this is refused
/// by name rather than converted — the OSM planet distribution publishes no sidecar digest
/// on this box, so a size floor is the strongest honest check available, and it catches the
/// failure that actually happens (a truncated or wrong file under the right name).
pub const MIN_DECLARED_PLANET_BYTES: u64 = 1 << 30;
// ─────────────────────────────────────────────────────────────────────────────
// The live run state (shared behind Arc<Mutex<_>>): phase + provenance + catalog.
// ─────────────────────────────────────────────────────────────────────────────
/// **How one step of the chain stood when the pass looked at it.** The vocabulary is
/// `korp::pipeline::StepState`'s, deliberately: the Collectors tab shows the same three
/// words the ⚙ map-source verifier and `korp-server --chain` show.
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StepState {
/// The step's artefacts were already COMPLETE — nothing to do, and the cut said so
/// rather than reproducing 30–50 minutes of work.
AlreadyComplete,
/// This pass produced them.
Produced,
/// This pass could not run the step — and names WHY in [`StepReport::note`]. The
/// honest answer for d–g in a build with no [`CutDriver`] injected; never a silent
/// skip that reads as done.
Unavailable,
/// The step ran and FAILED. The reason is in [`StepReport::note`].
Failed,
}
impl StepState {
/// The word the matrix cell and the log print.
pub fn label(self) -> &'static str {
match self {
StepState::AlreadyComplete => "already complete",
StepState::Produced => "produced",
StepState::Unavailable => "unavailable",
StepState::Failed => "failed",
}
}
/// Did the step end with its artefacts on disk (however they got there)?
pub fn is_complete(self) -> bool {
matches!(self, StepState::AlreadyComplete | StepState::Produced)
}
}
/// **What one step of one region's cut did** — the applied output, measured, never a
/// state round-trip. This is the row the Collectors tab renders and the row that lands
/// in [`SUMMARY_TABLE`].
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StepReport {
/// The step's letter, `a`..`g` — the SAME letter `korp::pipeline::Step::letter`
/// publishes and `korp-server --invalidate <region>:<letter>` accepts.
pub letter: char,
/// The step's key (`planet`, `coords`, … `address`).
pub step: String,
/// How it stood.
pub state: StepState,
/// Rows the step's defining artefact holds, when the step has a row count at all
/// (a directory of networks does not; `nodes.parquet` does). `0` is honest for a
/// step whose artefact is not a table.
pub rows: u64,
/// Bytes on disk the step's artefacts occupy.
pub bytes: u64,
/// Wall-clock milliseconds this pass spent on the step. `0` for a step it did not
/// have to run.
pub wall_ms: u64,
/// The one sentence an operator reads: why it was unavailable, what it produced,
/// what failed.
pub note: String,
}
impl StepReport {
/// The operator's line — `e networks: already complete (1.2 GB)`.
pub fn line(&self) -> String {
format!(
"{} {}: {} ({})",
self.letter,
self.step,
self.state.label(),
self.note
)
}
}
/// **The verdict for one region's whole chain.**
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RegionCutReport {
pub region: String,
pub steps: Vec<StepReport>,
/// **Post-conditions that MEASURED but could not ASSERT.** A pass whose only
/// geographic claim was "nothing was declared" is not green on the strength of a
/// check that cannot fail — nine such guards were found on this box in one day. The
/// steps may all be complete and the cut still be *impaired*, and this is where that
/// says so ([`crate::osm_cut_verify::CutVerdict::unasserted_reason`]).
pub caveats: Vec<String>,
/// **The GRAPH projection of this cut**, when the driver made one
/// ([`CutDriver::project_graph`]). Deliberately NOT one of `steps`: the letters
/// there are the contract with `korp::pipeline`, and the graph is a projection of
/// step e rather than a stage of the chain. `None` = not attempted.
pub graph: Option<StepReport>,
}
impl RegionCutReport {
/// Complete only when EVERY step of the declared range ended with its artefacts on
/// disk. An `Unavailable` step makes the cut incomplete — that is the whole reason
/// the variant exists.
pub fn complete(&self) -> bool {
!self.steps.is_empty() && self.steps.iter().all(|s| s.state.is_complete())
}
/// The first step that is not complete, for the phase/log line.
pub fn first_gap(&self) -> Option<&StepReport> {
self.steps.iter().find(|s| !s.state.is_complete())
}
/// The one-line verdict the Collectors tab shows per region.
pub fn verdict(&self) -> String {
if let Some(s) = self.first_gap() {
return format!("{}: incomplete at {}", self.region, s.line());
}
if let Some(g) = self.graph.as_ref().filter(|g| !g.state.is_complete()) {
return format!("{}: complete a–g but the GRAPH {}", self.region, g.line());
}
if !self.caveats.is_empty() {
return format!(
"{}: complete but UNASSERTED — {}",
self.region,
self.caveats.join("; ")
);
}
match &self.graph {
Some(g) => format!("{}: COMPLETE a–g, {}", self.region, g.line()),
None => format!("{}: COMPLETE a–g", self.region),
}
}
/// Green only when every step landed, every post-condition could actually assert,
/// and any graph projection that was ATTEMPTED succeeded. A
/// measured-but-unassertable check is impaired, not passed; so is a graph the cut
/// tried and failed to load.
pub fn green(&self) -> bool {
self.complete()
&& self.caveats.is_empty()
&& self.graph.as_ref().is_none_or(|g| g.state.is_complete())
}
}
// ─────────────────────────────────────────────────────────────────────────────
// CutDriver — the seam korp injects steps d–g through (LAW 5: one writer).
// ─────────────────────────────────────────────────────────────────────────────
/// **Who runs the steps this crate cannot.**
///
/// Steps d (relation clip + geometry assembly), e (networks + adjacency graph), f (the
/// packed base table) and g (the address index) have exactly ONE implementation in the
/// house, and it lives in **korp** — `regions::start_or_resume` / `pipeline`, the very
/// chain korp-server drives at boot. This crate is korp-free by charter, so it cannot
/// call it and **must not reimplement it**: a second chain would be the twin LAW 5
/// exists to forbid, and the two copies would disagree the first time either moved.
///
/// So korp implements this trait over its own chain and hands it to the daemon. A build
/// with no driver (the standalone `korp-collectors` binary, a leaf test) reports d–g as
/// [`StepState::Unavailable`] naming the reason, which is the honest answer — not a
/// skipped step that reads as done.
pub trait CutDriver: Send + Sync {
/// A short name for the implementation, for the log (`korp::regions`).
fn describe(&self) -> String;
/// **Bring `region`'s chain up through `step`, in `work_dir`.** Idempotent and
/// restart-safe: a step whose artefacts are already complete must return
/// [`StepState::AlreadyComplete`] and do no work — that is what lets a refresh tick
/// run against a live 30–50-minute cut without re-cutting it.
///
/// `stop` is the daemon's cooperative shutdown. An implementation checks it on
/// entry (a stopped daemon runs nothing) and passes it into any engine that has
/// a cancellation seam (osm-katana's `CancelToken` checkpoints at 1–15 ms), so a
/// `systemctl stop` during a long step reaches the work instead of waiting out
/// the step. A step abandoned on stop reports [`StepState::Unavailable`] with the
/// reason — never `Failed`, a stop is not a defect.
fn run_step(
&self,
region: &str,
work_dir: &Path,
step: state::Step,
log: &crate::daemon::LogSink,
stop: &crate::daemon::Shutdown,
) -> StepReport;
/// **Project the finished cut into the GRAPH plane** — the last thing a cut does.
///
/// The owner, 2026-09-02: *"and also it should put stuff in falkordb, ok?!"* Every
/// other collector in the object plane lands on both sides (korp-v2: collectors →
/// tables AND graph), and this one was landing on neither. Step e already produces
/// the road/rail adjacency (`networks/graph_edges.parquet` + `graph_nodes.parquet`);
/// this puts a BOUNDED slice of it into the configured graph row so it shows up in
/// the 🕸 Graph explorer beside the others — listed by the BACKEND, never injected
/// by a client.
///
/// It is **not** a step of the `a..g` chain and must never be given a letter from
/// it: the letters are the contract with `korp::pipeline`, and the graph is a
/// projection OF step e, not a stage after step g.
///
/// `None` = this build does not project graphs (the honest answer for a leaf-only
/// build; the collector then reports nothing rather than a green over an absent
/// graph). The default is `None`.
///
/// # BOUNDED is not optional
///
/// The full Sweden adjacency measures **3 694 448 nodes · 4 689 310 edges**, and a
/// 121 M-element load OOM-KILLED valkey at 16.7 GB on 2026-09-01. An implementation
/// MUST refuse before loading when the projection exceeds the store's cgroup
/// `memory.max`, naming the numbers.
fn project_graph(
&self,
_region: &str,
_work_dir: &Path,
_declaration: &serde_json::Value,
_log: &crate::daemon::LogSink,
) -> Option<StepReport> {
None
}
}
/// The driver used when korp has injected none — d–g are reported `Unavailable` with the
/// reason, and nothing is silently skipped.
pub struct NoDriver;
impl CutDriver for NoDriver {
fn describe(&self) -> String {
"none (leaf-only build)".into()
}
fn run_step(
&self,
region: &str,
_work_dir: &Path,
step: state::Step,
_log: &crate::daemon::LogSink,
_stop: &crate::daemon::Shutdown,
) -> StepReport {
StepReport {
letter: step.letter(),
step: step.label().into(),
state: StepState::Unavailable,
rows: 0,
bytes: 0,
wall_ms: 0,
note: format!(
"region '{region}': step {} ({}) has ONE implementation and it lives in korp \
(`regions::start_or_resume`); this build injected no CutDriver, so the step \
was NOT run and the cut is NOT complete",
step.letter(),
step.description(),
),
}
}
}
/// The mutable inner run state — the daemon thread updates it; the matrix reads it.
#[derive(Clone, Debug, Default)]
pub struct OsmCutRun {
pub phase: Phase,
/// The region the pass is on.
pub region: String,
/// The step letter the pass is on (`'\0'` before the first step).
pub step: char,
/// The last full verdict per configured region — what the Collectors tab renders.
pub reports: Vec<RegionCutReport>,
pub last_run_unix: u64,
/// The region CATALOG enumerated from the planet (design §7) — the pick-list.
pub catalog: Vec<String>,
pub error: Option<String>,
}
// ─────────────────────────────────────────────────────────────────────────────
// OsmCutDaemon — Config + State + the run loop.
// ─────────────────────────────────────────────────────────────────────────────
/// The `osm-cut` daemon. Cheap to clone the handle (shares the run state).
pub struct OsmCutDaemon {
enabled: bool,
config: serde_json::Value,
run: Arc<Mutex<OsmCutRun>>,
/// Who runs steps d–g ([`CutDriver`]). [`NoDriver`] until korp injects its chain.
driver: Arc<dyn CutDriver>,
}
impl Default for OsmCutDaemon {
fn default() -> Self {
Self::new()
}
}
impl OsmCutDaemon {
/// A disabled daemon with default config (the registry entry; korp overlays the
/// persisted `enabled` + custom config before starting it).
pub fn new() -> Self {
Self {
enabled: false,
config: default_config(),
run: Arc::new(Mutex::new(OsmCutRun::default())),
driver: Arc::new(NoDriver),
}
}
/// A daemon with an explicit `enabled` + custom config (what korp builds from the
/// persisted settings before spawning [`Collector::run`]).
pub fn with_config(enabled: bool, config: serde_json::Value) -> Self {
Self {
enabled,
config,
run: Arc::new(Mutex::new(OsmCutRun::default())),
driver: Arc::new(NoDriver),
}
}
/// **Inject the implementation of steps d–g** — korp's own chain. Builder-style so
/// the registry entry stays a plain `new()`; korp calls this at spawn time.
pub fn driven_by(mut self, driver: Arc<dyn CutDriver>) -> Self {
self.driver = driver;
self
}
/// Who is running d–g in this build.
pub fn driver_name(&self) -> String {
self.driver.describe()
}
/// **The last step the declaration asks for** (`through_step`, default
/// [`default_through_step`] — the END of the chain). An operator who only wants the
/// tables can stop at `c`; the steps after it are then not run and not reported,
/// rather than reported missing.
pub fn through_step(&self) -> state::Step {
self.config
.get("through_step")
.and_then(|v| v.as_str())
.and_then(state::Step::parse)
.unwrap_or_else(default_through_step)
}
/// **Where the cut lands** — the work root holding one directory per region
/// (`<work_dir>/<region>/ways.parquet`, …). Defaults to korp's own
/// `$KORP_HOME/work` so the collector and korp-server cut into the SAME place; a
/// declaration may point it elsewhere (a scratch cut that must not touch the live
/// one).
pub fn work_dir(&self) -> PathBuf {
if let Some(d) = self.config.get("work_dir").and_then(|v| v.as_str()) {
if !d.trim().is_empty() {
return PathBuf::from(d);
}
}
default_work_dir()
}
/// Read the live run snapshot.
pub fn snapshot(&self) -> OsmCutRun {
self.run.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
fn update(&self, f: impl FnOnce(&mut OsmCutRun)) {
let mut g = self.run.lock().unwrap_or_else(|e| e.into_inner());
f(&mut g);
}
/// The configured regions (multi-select from the catalog); defaults to `["sweden"]`.
fn regions(&self) -> Vec<String> {
self.config
.get("regions")
.and_then(|v| v.as_array())
.map(|a| {
a.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect::<Vec<_>>()
})
.filter(|v| !v.is_empty())
.unwrap_or_else(|| vec!["sweden".to_string()])
}
/// **One full reconcile pass** — the body of the daemon loop, exposed so tests
/// drive it deterministically without the never-ending wrapper.
///
/// Resolves the planet, indexes the region catalog, then drives **every configured
/// region's chain a→`through_step`** and lands **Online** (or a `Degraded`/`Failed`
/// fact). Restart-safe and, crucially, **idempotent against a live cut**: a step
/// whose artefacts already read complete reports `already complete` and costs
/// nothing, so a refresh tick against `~/.korp/work/sweden` does not spend 30–50
/// minutes re-cutting what the map is currently being served from.
pub async fn reconcile_once(&self, ctx: &CollectorCtx) -> Result<()> {
self.set(ctx, Phase::Offline);
// ── a. Resolve the planet — IF this pass needs one at all. ──────────────
// A region whose work dir already holds a COMPLETE chain needs no planet, and
// demanding one would strand a perfectly servable region in Degraded ("no
// planet") — or, worse, spend the pass scanning 176 GB to enumerate a catalog
// nothing needs.
std::fs::create_dir_all(&ctx.disk_dir).ok();
let work_root = self.work_dir();
std::fs::create_dir_all(&work_root).ok();
let regions = self.regions();
let last = self.through_step();
ctx.log.line(format!(
"cut plan: regions [{}] → {} through step {} ({}), driver = {}",
regions.join(", "),
work_root.display(),
last.letter(),
last.description(),
self.driver.describe(),
));
let planet_needed = regions.iter().any(|r| {
!declared_extract_complete(&self.config, r)
&& !extract_complete(®ion_work_dir(&work_root, r))
});
let planet = if planet_needed {
match self.resolve_planet(ctx).await {
Ok(p) => {
ctx.log.line(format!("planet resolved: {}", p.display()));
Some(p)
}
Err(e) => {
let reason = format!("no planet: {e:#}");
ctx.log.line(&reason);
self.set(ctx, Phase::Degraded { reason });
return Ok(()); // a missing planet is a FACT (degraded), not a hard failure
}
}
} else {
// No CONVERT needs a planet this pass. But if one is already sitting there —
// declared and readable, or recorded complete under the collector's own dir —
// the region catalog (the ⚙ Settings pick-list) is still worth refreshing off
// it, and `index_catalog` is cached on the planet's identity so it costs one
// read. What is skipped is ACQUISITION: nothing is downloaded, linked or
// torrented for a pass that has nothing to convert.
let at_hand = self
.config
.get("planet")
.and_then(|v| v.as_str())
.map(PathBuf::from)
.filter(|p| p.is_file())
.or_else(|| cached_planet(&ctx.disk_dir));
match &at_hand {
Some(p) => ctx.log.line(format!(
"every configured region ({}) already holds a complete pass-2 extract — no \
convert this pass; the planet already at {} is only read to refresh the \
region catalog",
regions.join(", "),
p.display(),
)),
None => ctx.log.line(format!(
"every configured region ({}) already holds a complete pass-2 extract — the \
planet is not needed this pass (no acquisition, no catalog scan)",
regions.join(", "),
)),
}
at_hand
};
// ── The region catalog from the planet (design §7) — the pick-list. ─────
self.set(ctx, Phase::Indexing);
// An enumeration FAILURE must not be laundered into an empty catalog. An
// empty catalog is a documented FACT (a fixture with no admin relations),
// so `unwrap_or_default()` here made a broken planet scan indistinguishable
// from a legitimately region-less planet: both logged "0 region(s)" and both
// reached ONLINE. Carry the failure to the verdict instead — the RAGNARÖK law
// is that landing nothing is RED unless the reason is stated.
let mut catalog_failure: Option<String> = None;
let catalog = match &planet {
Some(planet) => match index_catalog(
planet,
&ctx.disk_dir,
DigestMode::from_json(self.config.get("planet_digest")),
) {
Ok(c) => c,
Err(e) => {
let reason = format!("region catalog: {e:#}");
ctx.log.line(&reason);
catalog_failure = Some(reason);
Vec::new()
}
},
// No planet this pass ⇒ no scan. The previously cached catalog (if any) is
// still the truth about what the planet held; an absent one is an empty
// PICK-LIST, not a failed enumeration, and is reported as such.
None => cached_catalog(&ctx.disk_dir).unwrap_or_default(),
};
self.update(|r| r.catalog = catalog.clone());
if catalog_failure.is_none() {
ctx.log
.line(format!("region catalog: {} region(s)", catalog.len()));
}
// ── b..g. Drive every configured region's chain. ────────────────────────
let mut reports: Vec<RegionCutReport> = Vec::new();
for region in regions.iter().cloned() {
if ctx.shutdown.is_stopped() {
break;
}
self.update(|r| r.region = region.clone());
let report = self.cut_one_region(ctx, ®ion, &work_root, planet.as_deref(), last);
for s in report.steps.iter().chain(report.graph.iter()) {
ctx.log.line(format!("region '{region}': {}", s.line()));
self.land_step_summary(ctx, ®ion, s);
}
ctx.log.line(report.verdict());
reports.push(report);
}
// ── The verdict. ───────────────────────────────────────────────────────
let incomplete: Vec<String> = reports
.iter()
.filter(|r| !r.green())
.map(|r| r.verdict())
.collect();
let reason = catalog_failure.clone().or_else(|| {
if incomplete.is_empty() {
None
} else {
Some(incomplete.join("; "))
}
});
self.update(|r| {
r.reports = reports.clone();
r.last_run_unix = now_unix();
r.error = reason.clone();
});
// **A step that FAILED is a failure, not an impairment.** A post-condition that
// refused the cut, a convert that died, an unreadable extract — those are
// `Phase::Failed` and an `Err` out of the pass, because the daemon loop's retry
// cadence (60 s) is the right answer to them. An `Unavailable` step is different:
// nothing is broken, this build simply cannot run it, and retrying every minute
// would say nothing new. That one is Degraded.
if let Some(f) = reports
.iter()
.flat_map(|r| r.steps.iter().map(move |s| (r, s)))
.find(|(_, s)| s.state == StepState::Failed)
{
let why = format!("region '{}': {}", f.0.region, f.1.line());
self.update(|r| {
r.reports = reports.clone();
r.last_run_unix = now_unix();
r.error = Some(why.clone());
});
self.set(ctx, Phase::Failed { reason: why.clone() });
ctx.log.line(format!("FAILED — {why}"));
anyhow::bail!("{why}");
}
match &reason {
// Real work may have been done — but a chain with a hole in it is impaired,
// not green. Reporting ONLINE here would re-hide the very gap the step
// reports exist to show.
Some(reason) => {
self.set(
ctx,
Phase::Degraded {
reason: reason.clone(),
},
);
ctx.log
.line(format!("DEGRADED — {} region(s): {reason}", reports.len()));
}
None => {
self.set(ctx, Phase::Online);
ctx.log.line(format!(
"ONLINE — {} region(s) complete a–{}",
reports.len(),
last.letter(),
));
}
}
Ok(())
}
/// **One region's chain, a→`last`.** Steps b and c are osm-katana's convert, driven
/// here with the region's clip; d–g go through the injected [`CutDriver`], which is
/// korp's own chain. Step a is the planet the caller already resolved.
fn cut_one_region(
&self,
ctx: &CollectorCtx,
region: &str,
work_root: &Path,
planet: Option<&Path>,
last: state::Step,
) -> RegionCutReport {
let work = region_work_dir(work_root, region);
std::fs::create_dir_all(&work).ok();
let mut out = RegionCutReport {
region: region.to_string(),
steps: Vec::new(),
caveats: Vec::new(),
graph: None,
};
// **Where the extract actually IS.** Normally `work/<region>`, but a
// declaration may point `extract_dir` at an extract korp's own provisioner cut
// somewhere else — and step c must then read THAT file, not an absent one in
// `work`. Steps d–g still run against `work`: korp's chain is rooted there.
let mut src = work.clone();
for step in state::Step::ALL.into_iter().filter(|s| *s <= last) {
if ctx.shutdown.is_stopped() {
break;
}
self.set_step(ctx, region, step, 0.0);
let t0 = std::time::Instant::now();
let rep = match step {
// a — the planet. Already resolved (or honestly absent) above.
state::Step::Planet => match planet {
Some(p) => StepReport {
letter: 'a',
step: step.label().into(),
state: StepState::AlreadyComplete,
rows: 0,
bytes: std::fs::metadata(p).map(|m| m.len()).unwrap_or(0),
wall_ms: 0,
note: format!("planet {}", p.display()),
},
None => StepReport {
letter: 'a',
step: step.label().into(),
state: StepState::AlreadyComplete,
rows: 0,
bytes: 0,
wall_ms: 0,
note: "no planet needed — the extract downstream is already complete"
.into(),
},
},
// b + c — osm-katana's two convert passes, run TOGETHER (one scan
// produces both `nodes.parquet`/`node_coords.arrow` and
// `ways.parquet`/`relations.parquet`; splitting them here would mean
// scanning the planet twice). Step b reports the convert; step c reads
// the artefact the same convert produced.
state::Step::Coords => {
let (rep, caveat, resolved) =
self.convert_step(ctx, region, &work, planet, step);
out.caveats.extend(caveat);
src = resolved;
rep
}
state::Step::Tables => step_from_disk(step, &src, &["ways.parquet"]),
// d–g — korp's chain, through the seam.
_ => self.driver.run_step(region, &work, step, &ctx.log, &ctx.shutdown),
};
let mut rep = rep;
if rep.wall_ms == 0 {
rep.wall_ms = t0.elapsed().as_millis() as u64;
}
// Per-step job history (ctx.jobs): only work that RAN lands a row —
// Produced and Failed. AlreadyComplete ticks record nothing, or an
// hourly refresh would write nine confirmations per region per tick.
if matches!(rep.state, StepState::Produced | StepState::Failed) {
ctx.jobs.record(
&format!("{region}:{}", rep.letter),
rep.state == StepState::Produced,
rep.wall_ms,
serde_json::json!({
"step": rep.step,
"rows": rep.rows,
"bytes": rep.bytes,
"note": rep.note,
}),
);
}
let stop = rep.state == StepState::Failed;
out.steps.push(rep);
if stop {
// A failed step makes every step after it meaningless — say so once
// rather than reporting six more failures with the same cause.
return out;
}
}
// ── The GRAPH projection — the last thing a cut does, and only over a chain
// that actually finished. Projecting a half-cut region would load an
// adjacency nothing vouches for.
if out.complete() && !ctx.shutdown.is_stopped() {
out.graph = self.driver.project_graph(region, &work, &self.config, &ctx.log);
}
out
}
/// **Steps b/c — convert the planet into `work/<region>` with the region's clip.**
/// Reuses the extract that is already there when it reads complete: the whole point
/// of the rewrite is that a refresh tick against the LIVE sweden cut costs nothing.
fn convert_step(
&self,
ctx: &CollectorCtx,
region: &str,
work: &Path,
planet: Option<&Path>,
step: state::Step,
) -> (StepReport, Option<String>, PathBuf) {
let mk = |state: StepState, note: String, bytes: u64, rows: u64| StepReport {
letter: step.letter(),
step: step.label().into(),
state,
rows,
bytes,
wall_ms: 0,
note,
};
if extract_complete(work) {
let (rows, bytes) = table_rows_and_bytes(&work.join("nodes.parquet"));
return (
mk(
StepState::AlreadyComplete,
format!("{} holds a complete pass-1/pass-2 extract", work.display()),
bytes,
rows,
),
None,
work.to_path_buf(),
);
}
// A DECLARED `extract_dir` that reads complete is cut FROM, planet or no planet
// — that is the whole point of declaring one, and `resolve_cut_extract` below is
// where it is picked up. So the planet is only demanded when there is nothing
// else to cut from.
let declared = declared_extract_complete(&self.config, region);
let planet_path;
let planet = match (planet, declared) {
(Some(p), _) => p,
(None, true) => {
planet_path = PathBuf::new();
planet_path.as_path()
}
(None, false) => {
return (
mk(
StepState::Failed,
format!(
"{} holds no complete extract, no extract_dir is declared, and no \
planet was resolved — nothing to convert FROM",
work.display()
),
0,
0,
),
None,
work.to_path_buf(),
);
}
};
let clip = resolve_region_cut_clip(&self.config, region, planet, &ctx.log);
if clip.is_none() {
ctx.log.line(format!(
"region '{region}': no known clip — converting the WHOLE source as this region \
(correct only for a fixture / an already-regional extract)"
));
}
let t0 = std::time::Instant::now();
let mut plan = RegionCut::from_config(&self.config, planet, region, work, clip);
let stop = ctx.shutdown.clone();
plan.stop = Some(Arc::new(move || stop.is_stopped()));
match cut_region_extract(&plan, &ctx.log) {
Ok(s) => {
let caveat = crate::osm_cut_verify::CutVerdict {
checks: s.checks.clone(),
}
.unasserted_reason()
.map(|r| format!("region '{region}': {r}"));
let (rows, bytes) = table_rows_and_bytes(&s.extract_dir.join("nodes.parquet"));
let mut r = mk(
if s.reused_extract {
StepState::AlreadyComplete
} else {
StepState::Produced
},
format!(
"{} → {}",
if s.reused_extract {
"reused the extract in place"
} else {
"converted the planet"
},
s.extract_dir.display()
),
bytes,
rows,
);
r.wall_ms = t0.elapsed().as_millis() as u64;
let resolved = s.extract_dir.clone();
(r, caveat, resolved)
}
Err(e) => (
mk(StepState::Failed, format!("{e:#}"), 0, 0),
None,
work.to_path_buf(),
),
}
}
/// Land one step's SUMMARY row through the ordinary warehouse seam. Best-effort and
/// LOUD on failure: a receipt that could not be written is worth saying out loud,
/// and is never worth failing a finished cut over.
fn land_step_summary(&self, ctx: &CollectorCtx, region: &str, s: &StepReport) {
match step_summary_batch(region, s) {
Ok(batch) => {
if let Err(e) = ctx.warehouse.append(SUMMARY_TABLE, &batch) {
ctx.log.line(format!(
"region '{region}': the {} summary row did not land in {SUMMARY_TABLE}: \
{e:#} (the cut itself is unaffected — it writes FILES)",
s.letter
));
}
}
Err(e) => ctx
.log
.line(format!("region '{region}': summary row not built: {e:#}")),
}
}
/// Report the phase for a step, with its letter — what the Collectors tab shows.
fn set_step(&self, ctx: &CollectorCtx, region: &str, step: state::Step, pct: f32) {
self.update(|r| {
r.region = region.to_string();
r.step = step.letter();
});
self.set(
ctx,
Phase::Cutting {
region: format!("{region} {} {}", step.letter(), step.label()),
pct,
},
);
}
/// Resolve the planet path: an explicit `config.planet` (the seed / test
/// fixture), else a cached `*.osm.{pbf,bz2}` under `disk_dir`, else the well-known
/// `$HOME/planet-260427.osm.bz2` seed (symlinked into the cache), else — under
/// `osm-daemon-download` — the librqbit torrent. The download lives HERE (§9).
async fn resolve_planet(&self, ctx: &CollectorCtx) -> Result<PathBuf> {
// 1) explicit path in config (the test/seed handoff).
//
// A DECLARED planet is a producer too: the operator (or a test) asserting "this
// file is a planet" is the completion event, so it is recorded like any other
// rather than being trusted afresh on every pass. `is_file()` is all this arm can
// check — that is precisely why the fact gets written down instead of re-derived.
if let Some(p) = self.config.get("planet").and_then(|v| v.as_str()) {
let p = PathBuf::from(p);
if p.is_file() {
if let Err(e) = record_planet(
&ctx.disk_dir,
&p,
&format!("declared config.planet = {}", p.display()),
"operator declaration (config.planet)",
) {
ctx.log
.line(format!("could not record the declared planet: {e:#}"));
}
return Ok(p);
}
}
// 2) already recorded complete under the collector dir.
if let Some(p) = cached_planet_reporting(&ctx.disk_dir, &ctx.log) {
return Ok(p);
}
// 3) the operator's DECLARED seed paths, linked in through the Acquire seam.
//
// This arm used to read `$HOME/planet-260427.osm.bz2`. That cost twice: the answer
// to "where does this collector get its planet" depended on whichever account
// happened to run the process rather than on anything reviewable, and because the
// arm always hit on the developer's box the torrent behind it was never exercised.
// A declared path is in the config, an operator can read it, and a service account
// that has no such file gets a refusal naming every path tried.
if let Some(planet) = self.seeded_planet(ctx) {
return Ok(planet);
}
// 4) torrent (feature-gated so the lean build pulls no librqbit).
self.torrent_planet(ctx).await
}
/// The declared seed paths, in order, as `config.seeds` (a JSON array) or the single
/// `config.seed`. Empty when the operator declared none — and then this arm simply does
/// not apply, which is what lets the transport behind it be reached and tested.
fn declared_seeds(&self) -> Vec<PathBuf> {
let mut out = Vec::new();
if let Some(one) = self.config.get("seed").and_then(|v| v.as_str()) {
if !one.trim().is_empty() {
out.push(PathBuf::from(one));
}
}
if let Some(list) = self.config.get("seeds").and_then(|v| v.as_array()) {
out.extend(
list.iter()
.filter_map(|v| v.as_str())
.filter(|s| !s.trim().is_empty())
.map(PathBuf::from),
);
}
out
}
/// Link a declared seed into the collector's own directory and RECORD it complete.
///
/// The link (never a copy — 176 GB) and the completeness check both come from
/// [`crate::acquire::LocalSeed`], so a seed and an HTTP fetch land under the same rules.
/// `None` when no declared seed applied; the reason is logged, because a seed the
/// operator declared and this refused is exactly what they need to see.
fn seeded_planet(&self, ctx: &CollectorCtx) -> Option<PathBuf> {
use crate::acquire::{Acquire, AcquireRequest, AcquireSink, Completeness, LocalSeed};
let declared = self.declared_seeds();
if declared.is_empty() {
return None;
}
let dest = ctx.disk_dir.join(
declared[0]
.file_name()
.unwrap_or(std::ffi::OsStr::new(DEFAULT_SEED)),
);
let req = AcquireRequest::local_seed(declared, &dest)
// A planet is at minimum a gigabyte; anything shorter is not one, whatever it
// is called. The honest check for a distribution that publishes no digest.
.requiring(Completeness::MinBytes(MIN_DECLARED_PLANET_BYTES))
.cancelled_by(ctx.shutdown.clone());
match LocalSeed::linking().acquire(&req, &AcquireSink::quiet(ctx.log.clone())) {
Ok(got) => {
self.set(ctx, Phase::Downloading { pct: 1.0 });
if let Err(e) = record_planet(
&ctx.disk_dir,
&got.path,
&format!("declared seed {} ({})", got.path.display(), got.checked),
"operator-declared seed (acquire::LocalSeed)",
) {
ctx.log
.line(format!("could not record the seeded planet: {e:#}"));
}
Some(got.path)
}
Err(e) => {
ctx.log
.line(format!("declared seed not usable, falling through: {e:#}"));
None
}
}
}
/// **Torrent the planet OFF the async worker.** osm-katana's
/// [`osm_katana::fixture::ensure_planet_osm`] builds its OWN tokio runtime and
/// `block_on`s it. Called inline on korp-server's tokio worker (which drives this
/// `async fn`) that nests a runtime inside a runtime and PANICS with "cannot
/// start a runtime from within a runtime" — so the librqbit branch never actually
/// executed through the daemon. We move the blocking acquisition onto a
/// `spawn_blocking` thread, which carries NO tokio runtime context, so
/// osm-katana's runtime is top-level. FIVE ZEROS: no busy-wait; the blocking
/// torrent leaves the async worker free.
#[cfg(feature = "osm-daemon-download")]
async fn torrent_planet(&self, ctx: &CollectorCtx) -> Result<PathBuf> {
self.set(ctx, Phase::Downloading { pct: 0.0 });
ctx.log.line("torrenting the OSM planet (librqbit)…");
let planet = tokio::task::spawn_blocking(osm_katana::fixture::ensure_planet_osm)
.await
.context("planet torrent task join")?
.context("planet torrent download")?;
// ── THE COMPLETION MOMENT ────────────────────────────────────────────
// `ensure_planet_osm` returning `Ok` IS librqbit saying the torrent finished.
// That is the fact, and it is written down HERE — not re-derived later from
// block allocation, which is a guess about this very event.
//
// `seal` fsyncs the planet before committing the row, so a crash in between
// leaves no record and the next pass re-acquires, rather than a durable claim
// about a file that never landed.
record_planet(
&ctx.disk_dir,
&planet,
&format!("torrent {}", planet.display()),
"librqbit (osm_katana::fixture::ensure_planet_osm)",
)
.with_context(|| {
format!(
"the torrent completed but its completion could not be RECORDED — refusing \
to report a planet whose completeness nothing vouches for ({})",
planet.display()
)
})?;
ctx.log.line(format!(
"planet acquired and RECORDED complete: {} (librqbit)",
planet.display()
));
Ok(planet)
}
#[cfg(not(feature = "osm-daemon-download"))]
async fn torrent_planet(&self, _ctx: &CollectorCtx) -> Result<PathBuf> {
anyhow::bail!(
"no planet on disk and the torrent fetcher is not compiled in — DECLARE a seed path \
in this collector's config (`\"seed\": \"/path/to/planet-*.osm.bz2\"`, or `\"seeds\"` \
for several tried in order), or build with --features osm-daemon-download. No \
directory is searched: a declared path is reviewable, an ambient one is whatever \
the account running this process happened to have"
)
}
/// Set the phase both in the run state and out through the ctx (jera + BIOS).
fn set(&self, ctx: &CollectorCtx, phase: Phase) {
self.update(|r| r.phase = phase.clone());
ctx.set_phase(&phase);
}
}
impl CollectorConfig for OsmCutDaemon {
fn name(&self) -> &str {
NAME
}
/// **The graph osm-cut's two landed tables honestly make** — lane GQ2.
///
/// Until now osm-cut was the one collector that wrote its own graph, through a
/// direct `GRAPH.BULK` call, because it implements the DAEMON trait and only the
/// ingest trait could declare a mapping. It had nowhere to say what its columns
/// mean. This is that place, and it makes the road graph an ordinary pipeline: a
/// plan in the Pipelines explorer with a last run, a freshness and a ▶ Run, like
/// every other collector's graph.
///
/// The columns are `osm_katana::graph::{node_fields, edge_fields}` verbatim —
/// checked against the writer, not remembered:
///
/// * `graph_nodes` — `id` (the OSM node id, kept so the join back to `rail_nodes`
/// and to OSM is an exact `i64`), `kind` (`road` · `rail` · `both`, where `both`
/// IS the level-crossing set by construction), `degree`, `rail_class`.
/// * `graph_edges` — `from_node` / `to_node` as the endpoints, carrying `network`,
/// `class`, `speed_kmh`, `oneway` and `length_m`. All four are resolved at BUILD
/// time by the cut precisely so a router touching every edge it settles never
/// re-parses a tag to price it; putting them on the relationship keeps that true
/// in the graph store too.
///
/// # Two things this does NOT declare, and why
///
/// **No `lon`/`lat` on the Junction.** `graph_nodes` has no such columns — the
/// position lives in the WKB `geometry` field. Declaring them would name columns
/// the collector does not emit, which is the exact defect lane EC O3 found in 29 of
/// 53 business keys. Giving a Junction a real position needs either `geometry` as a
/// property (WKB bytes into FalkorDB) or lon/lat columns added to osm-katana's
/// `node_fields` and the cut re-run; both are decisions, not details.
///
/// **One relationship type, not `ROAD` and `RAIL`.** The road/rail split lives in
/// the `network` COLUMN, and a declaration has no row predicate — two `EdgeDecl`s
/// over one table would each emit every edge, so a `RAIL` type would also contain
/// every motorway. `network` rides as a property instead, which is askable in
/// Cypher and honest; a real type split needs a filter in the declaration.
fn graph_mappings(&self) -> Vec<crate::graph_mapping::GraphMapping> {
use crate::graph_mapping::GraphMapping;
vec![
GraphMapping::new(
GRAPH_PLAN,
"🛣 Sweden roads + rails",
"graph_nodes",
"a Junction carries no coordinate: `graph_nodes` keeps its position in the \
WKB `geometry` column, not in lon/lat columns",
)
.vertex("Junction", "id", &["kind", "degree", "rail_class"]),
GraphMapping::new(
GRAPH_PLAN,
"🛣 Sweden roads + rails",
"graph_edges",
"`network` distinguishes road from rail as a PROPERTY, not as the \
relationship type — a declaration has no row predicate, so one type per \
network would put every motorway in both",
)
.edge_with(
"CONNECTS",
("Junction", "from_node", "id"),
("Junction", "to_node", "id"),
&["network", "class", "speed_kmh", "oneway", "length_m", "way_id"],
),
]
}
/// **osm-cut needs no URL, and saying it needs one would be a lie.** It resolves the
/// planet from DISK (`$KORP_PLANET` / its own cache dir, seeded by a torrent), so the
/// default derivation — "the `url` key of `defaults()`, empty ⇒ no source" — reads
/// `NeedsConfig` for a collector that is up and serving. It declares the planet as
/// its source instead, which is where its bytes actually come from.
///
/// It declares no required config field: with no region picked it still holds ONLINE
/// for the empty target set, and the pick-list's own "planet not downloaded yet" is
/// the honest thing to show before the catalog exists.
fn requirements(&self) -> crate::requires::Requirements {
crate::requires::Requirements::open(
"local:the on-disk planet ($KORP_PLANET, else the collector's own cache)",
)
}
fn enabled(&self) -> bool {
self.enabled
}
fn schema(&self) -> ConfigSchema {
// The region pick-list options are the live catalog (empty before the planet
// is indexed → Settings shows "planet not downloaded yet").
let catalog = self.snapshot().catalog;
ConfigSchema::new(vec![
ConfigField::of(
"regions",
"Regions",
"multi-select from the planet catalog",
FieldKind::RegionPickList { catalog },
),
ConfigField::text(
"planet",
"Planet file",
"path to planet-*.osm.{pbf,bz2} (blank ⇒ seed/torrent)",
),
ConfigField::text(
"work_dir",
"Cut into",
"the work ROOT holding one dir per region (blank ⇒ $KORP_HOME/work — the very source korp-server reads)",
),
ConfigField::text(
"through_step",
"Through step",
&steps_hint(),
),
ConfigField::text(
"clip_mode",
"Clip mode",
"auto (default: the region's own .poly, else its bbox) | polygon (derive from the planet) | bbox (coarse box)",
),
ConfigField::text(
"poly",
"Clip polygon",
"Osmosis .poly / GeoJSON text, or a path to one — the region's REAL shape (per-region map allowed)",
),
ConfigField::text(
"extract_dir",
"Reuse extract",
"an existing {nodes,ways}.parquet dir to cut from instead of re-converting the planet (per-region map allowed)",
),
ConfigField::text(
"revision",
"Cut revision",
"bump to RE-CUT (any string; part of the cut declaration)",
),
ConfigField::text(
"expect",
"Post-conditions",
"JSON the cut must SATISFY or the run FAILS: foreign_nodes/wholly_foreign_ways depth (off|sample:N|full), present[{name,lon,lat,radius_km,min_features,subject}], free_gb_convert",
),
ConfigField::text(
"keep_generations",
"Keep generations",
"how many past extracts to retain on disk (default 2; the OLDEST is pruned)",
),
ConfigField::text(
"planet_digest",
"Planet identity",
"sidecar (default: a published checksum beside the planet, else size+mtime) | blake3 (hash it — MEASURED 0.99 GiB/s ⇒ ~2.7 min for the 173 GB planet, cached per download) | size",
),
ConfigField::text(
"retry_secs",
"Retry interval",
"how often to retry while NOT online (default 60s; the refresh tick is for the online steady state)",
),
])
}
fn title(&self) -> String {
"🗺 osm-cut".to_string()
}
fn description(&self) -> String {
// The step letters are NOT restated here: a description that spells the chain is
// a second copy of it, and the one that goes stale silently because nothing
// parses prose. `steps_hint()` renders them from the ONE enum.
format!(
"Cut every configured region out of the OSM planet into the packed map source \
korp-server reads ({}). It writes FILES, not tables — the only rows it lands \
are one summary per region and step.",
steps_hint(),
)
}
fn defaults(&self) -> serde_json::Value {
serde_json::json!({
"regions": [],
"planet": "",
"work_dir": "",
"through_step": default_through_step_letter(),
"clip_mode": CLIP_MODE_DEFAULT,
"poly": "",
"extract_dir": "",
"keep_node_coords": true,
"regenerate": false,
"revision": "",
"keep_generations": 2,
"planet_digest": "sidecar",
"retry_secs": 60,
"expect": {},
})
}
/// **The real number, declared where it is known.** The planet alone was measured at
/// 176 366 896 594 B on 2026-08-04 (`.nornir/collector-runtime-design-2026-08-04.md`
/// §6.2), and the run then needs the node store, the extract parquet, the
/// networks and the packed base table beside it — so 300 GiB is the working set,
/// not the download.
///
/// It is a DECLARATION, not a placement rule. Nothing reads it and picks a
/// substrate; `UnitSpec::placement_advice` may warn and the programmer decides.
/// Before this, every unit korp built carried a flat `"20Gi"` — a number that was a
/// declaration for no collector at all and 8× too small for this one.
fn durable_bytes(&self) -> Option<u64> {
Some(300 * 1024 * 1024 * 1024)
}
}
impl CollectorState for OsmCutDaemon {
fn phase(&self) -> Phase {
self.snapshot().phase
}
fn facet(&self) -> serde_json::Value {
let s = self.snapshot();
serde_json::json!({
"collector": NAME,
"phase": s.phase.label(),
"state": s.phase,
"progress": s.phase.progress(),
"region": s.region,
"step": if s.step == '\0' { String::new() } else { s.step.to_string() },
"reports": s.reports,
"driver": self.driver.describe(),
"last_run_unix": s.last_run_unix,
"catalog": s.catalog,
"error": s.error,
})
}
}
#[async_trait::async_trait]
impl Collector for OsmCutDaemon {
async fn run(self: Arc<Self>, ctx: CollectorCtx) {
// The never-ending loop: reconcile toward ONLINE, then watch/refresh until
// asked to stop (the config toggle / server shutdown trips ctx.shutdown).
let refresh_secs = self.config.get("refresh_secs").and_then(|v| v.as_u64());
// **A daemon that is NOT online retries on a short cadence, not the refresh
// tick.** The owner, 2026-08-02: "den måste hantera detta som att den är i
// error=röd state och rapportera i sin fel log. och pröva om då då, typ 1 ggr
// minut." Waiting an hour to retry a transient fault is indistinguishable from
// having given up, and the matrix cell stays red the whole time.
let retry_secs = self
.config
.get("retry_secs")
.and_then(|v| v.as_u64())
.unwrap_or(60);
loop {
if let Err(e) = self.reconcile_once(&ctx).await {
ctx.log.line(format!(
"ERROR: reconcile failed (will retry in {retry_secs}s): {e:#}"
));
}
if ctx.shutdown.is_stopped() {
break;
}
// Idle-and-Online is the steady state of the same loop (design §2); a
// collector that is Degraded/Failed is NOT idling, it is waiting to try
// again, and it says so.
let healthy = self.phase().healthy();
if !healthy {
ctx.log.line(format!(
"ERROR: not online ({}) — retrying in {retry_secs}s",
self.phase().label()
));
}
let sleep = std::time::Duration::from_secs(if healthy {
refresh_secs.unwrap_or(3600)
} else {
retry_secs
});
let step = std::time::Duration::from_millis(200);
let mut waited = std::time::Duration::ZERO;
while waited < sleep {
if ctx.shutdown.is_stopped() {
break;
}
// A host-side poke (an invalidated step, a changed region config)
// ends the idle wait NOW — otherwise the purge sits unnoticed for
// up to refresh_secs.
if ctx.nudge.take() {
ctx.log.line("nudged — reconciling now instead of at the next tick");
break;
}
tokio::time::sleep(step).await;
waited += step;
}
if ctx.shutdown.is_stopped() {
break;
}
}
ctx.log.line("daemon stopped");
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Free helpers — the work dirs, the chain vocabulary, the planet cache and the catalog.
// ─────────────────────────────────────────────────────────────────────────────
/// **The default cut root — `$KORP_HOME/work`, else `~/.korp/work`.**
///
/// The same root korp-server verifies at boot (`server_regions::verify_map_source`) and
/// the same one the map is served from. The collector cuts INTO it rather than beside
/// it: two roots would mean the collector could report a complete chain the server
/// cannot see, which is exactly the shape the pyramid had.
pub fn default_work_dir() -> PathBuf {
if let Ok(h) = std::env::var("KORP_HOME") {
if !h.trim().is_empty() {
return PathBuf::from(h).join("work");
}
}
let home = std::env::var("HOME").unwrap_or_else(|_| ".".into());
PathBuf::from(home).join(".korp").join("work")
}
/// One region's directory under the cut root.
pub fn region_work_dir(work_root: &Path, region: &str) -> PathBuf {
work_root.join(region)
}
/// **The chain, rendered from the ONE enum** — the `through_step` placeholder and the
/// collector description both read this rather than spelling `a..g` in prose. A frozen
/// literal is how the leaf and korp drifted apart in the first place.
pub fn steps_hint() -> String {
state::Step::ALL
.iter()
.map(|s| format!("{} {}", s.letter(), s.label()))
.collect::<Vec<_>>()
.join(" · ")
}
/// **The default `through_step` — the END of the chain, read from the ONE enum.**
///
/// This was the frozen literal `g`, and it stopped being the end of the chain the day
/// `h` (landmass) and `i` (buildings) were appended to [`state::Step::ALL`]. MEASURED on
/// oden 2026-09-10: korp's `--invalidate` cascades to `i`, korp-server's resume walk
/// therefore printed `RESUME AT step h` — and the cut daemon, on its stock config, would
/// never run `h` or `i`, so **nothing on the box could resume it**. The chain announced a
/// step that had no producer. It was papered over as CONFIG (`through_step: "h"` in
/// `/home/korp/.config/korp/config.json`), which fixes one host and no other.
///
/// A default that is a letter has to be edited every time the chain grows, by whoever
/// remembers; a default that is `ALL.last()` extends itself. `j` is already coming.
///
/// An operator who wants less still says so — `through_step: "c"` is untouched. What is
/// no longer possible is a STOCK install whose last step is not the last step.
pub fn default_through_step() -> state::Step {
*state::Step::ALL.last().expect("the cut chain has at least one step")
}
/// [`default_through_step`] as the letter the config field carries (`"i"`).
pub fn default_through_step_letter() -> String {
default_through_step().letter().to_string()
}
/// Rows and bytes of a parquet artefact, for a [`StepReport`]. Both `0` when the file is
/// absent or unreadable — an honest "nothing measured", never a fabricated count.
pub fn table_rows_and_bytes(p: &Path) -> (u64, u64) {
let bytes = std::fs::metadata(p).map(|m| m.len()).unwrap_or(0);
(parquet_rows(p), bytes)
}
/// The parquet footer's row count, or `0` when the file is absent/footerless.
fn parquet_rows(p: &Path) -> u64 {
use parquet::file::reader::{FileReader, SerializedFileReader};
let Ok(f) = std::fs::File::open(p) else {
return 0;
};
match SerializedFileReader::new(f) {
Ok(r) => r.metadata().file_metadata().num_rows().max(0) as u64,
Err(_) => 0,
}
}
/// Bytes a whole directory tree occupies (the measure for a step whose artefact is a
/// directory, like `networks/`).
pub fn dir_bytes(p: &Path) -> u64 {
let Ok(rd) = std::fs::read_dir(p) else {
return 0;
};
let mut total = 0u64;
for e in rd.flatten() {
let Ok(md) = e.metadata() else { continue };
total += if md.is_dir() {
dir_bytes(&e.path())
} else {
md.len()
};
}
total
}
/// **A step whose completion is simply "these files are there and complete".** Used for
/// step c, whose artefacts the SAME convert as step b produced — running the convert
/// twice to report two steps would scan the planet twice for one answer.
pub fn step_from_disk(step: state::Step, work: &Path, files: &[&str]) -> StepReport {
let missing: Vec<&str> = files
.iter()
.copied()
.filter(|f| !parquet_complete(&work.join(f)))
.collect();
let (rows, bytes) = table_rows_and_bytes(&work.join(files[0]));
if missing.is_empty() {
StepReport {
letter: step.letter(),
step: step.label().into(),
state: StepState::AlreadyComplete,
rows,
bytes,
wall_ms: 0,
note: format!("{} in {}", files.join(" + "), work.display()),
}
} else {
StepReport {
letter: step.letter(),
step: step.label().into(),
state: StepState::Failed,
rows: 0,
bytes: 0,
wall_ms: 0,
note: format!(
"{} holds no complete {}",
work.display(),
missing.join(" / ")
),
}
}
}
/// **The one summary row a finished step lands** (see [`SUMMARY_TABLE`]).
///
/// The cut emits no map `RecordBatch` — it writes files. This is the receipt: what ran,
/// on what, how it stood, and what it cost. One row, one step, one region.
pub fn step_summary_batch(region: &str, s: &StepReport) -> Result<arrow_array::RecordBatch> {
use arrow_array::{RecordBatch, StringArray, UInt64Array};
use arrow_schema::{DataType, Field, Schema};
use std::sync::Arc as StdArc;
let schema = StdArc::new(Schema::new(vec![
Field::new("region", DataType::Utf8, false),
Field::new("step", DataType::Utf8, false),
Field::new("letter", DataType::Utf8, false),
Field::new("state", DataType::Utf8, false),
Field::new("rows", DataType::UInt64, false),
Field::new("bytes", DataType::UInt64, false),
Field::new("wall_ms", DataType::UInt64, false),
Field::new("note", DataType::Utf8, false),
Field::new("at_unix", DataType::UInt64, false),
]));
RecordBatch::try_new(
schema,
vec![
StdArc::new(StringArray::from(vec![region.to_string()])),
StdArc::new(StringArray::from(vec![s.step.clone()])),
StdArc::new(StringArray::from(vec![s.letter.to_string()])),
StdArc::new(StringArray::from(vec![s.state.label().to_string()])),
StdArc::new(UInt64Array::from(vec![s.rows])),
StdArc::new(UInt64Array::from(vec![s.bytes])),
StdArc::new(UInt64Array::from(vec![s.wall_ms])),
StdArc::new(StringArray::from(vec![s.note.clone()])),
StdArc::new(UInt64Array::from(vec![now_unix()])),
],
)
.context("build the osm-cut step summary row")
}
/// The default `osm-cut` config (the Settings defaults).
pub fn default_config() -> serde_json::Value {
serde_json::json!({
"regions": ["sweden"],
"planet": "",
"work_dir": "",
"through_step": default_through_step_letter(),
"clip_mode": CLIP_MODE_DEFAULT,
"poly": "",
"extract_dir": "",
"keep_node_coords": true,
"regenerate": false,
"revision": "",
})
}
/// **CORROBORATION ONLY — how many bytes of `md` the filesystem says it allocated**
/// (`st_blocks × 512`, capped at `st_size`).
///
/// # This is NOT the authority, and must never be made one again
///
/// It was, briefly, and that was wrong. Whether a download finished is a fact
/// **librqbit already knows**, and inferring it from block allocation is guessing at a
/// proxy that four unrelated things break: sparse files (a complete file reads short),
/// compressing filesystems (a complete file occupies fewer blocks than its size),
/// preallocation and delayed allocation (an *incomplete* file occupies more — the `min`
/// cap below exists solely to paper over that, and is the admission that this is a
/// correlation), and reflink copies (blocks are shared, so the accounting is not
/// per-file).
///
/// The authority is [`crate::state`]: the producer records completion at the moment its
/// own API says done. This function survives as the **second, independent** signal that
/// [`crate::state::judge`] corroborates a record against — two sources that can disagree
/// are worth more than one, and a disagreement is reported rather than resolved.
#[cfg(unix)]
pub fn bytes_on_disk(md: &std::fs::Metadata) -> u64 {
use std::os::unix::fs::MetadataExt;
md.blocks().saturating_mul(512).min(md.len())
}
#[cfg(not(unix))]
pub fn bytes_on_disk(md: &std::fs::Metadata) -> u64 {
md.len()
}
/// **CORROBORATION ONLY — does the filesystem think `p` looks whole?**
///
/// A heuristic, kept as the independent second opinion described on [`bytes_on_disk`].
/// **Do not call this to decide whether to cut** — call [`planet_standing`], which asks
/// the ledger and uses this only to corroborate the answer.
pub fn planet_looks_whole_on_disk(p: &Path) -> bool {
let Ok(md) = std::fs::metadata(p) else {
return false;
};
let total = md.len();
total > 0 && bytes_on_disk(&md) >= total
}
/// **THE AUTHORITY — what does the collector's own ledger say about the planet?**
///
/// Reads the [`state::Step::Planet`] record through a lock-free [`state::StateReader`]
/// (so it works while the daemon is writing) and judges it against the artefact the
/// record names. The four outcomes are all different and all actionable:
///
/// * [`state::Standing::Unrecorded`] — no producer ever declared a planet finished here.
/// Acquire one. This is the honest answer for an interrupted download, **however
/// complete the file happens to look**.
/// * [`state::Standing::Complete`] — recorded, and the file corroborates. Cut from it.
/// * [`state::Standing::Disagrees`] — recorded, but the file is gone or has changed size.
/// A **fault to report**, not a thing to quietly re-download: something outside this
/// collector is eating artefacts and hiding that helps nobody.
/// * [`state::Standing::Foreign`] — recorded from a different source. Refused by name.
pub fn planet_standing(disk_dir: &Path) -> state::Standing {
let reader = match state::StateReader::open(disk_dir) {
Ok(Some(r)) => r,
Ok(None) => {
return state::Standing::Unrecorded {
why: format!(
"no collector state store at {} — nothing has ever been recorded \
complete on this box",
disk_dir.display()
),
};
}
Err(e) => {
return state::Standing::Unrecorded {
why: format!(
"cannot read the collector state store at {}: {e:#}",
disk_dir.display()
),
};
}
};
let rec = match reader.get(state::Step::Planet, PLANET_KEY) {
Ok(r) => r,
Err(e) => {
return state::Standing::Unrecorded {
why: format!("cannot read the planet completion record: {e:#}"),
};
}
};
let artefact = rec
.as_ref()
.map(|r| PathBuf::from(&r.artefact))
.unwrap_or_else(|| disk_dir.join("planet"));
state::judge(rec, &artefact, None, None)
}
/// The ledger key every planet record is filed under. One planet per collector dir.
pub const PLANET_KEY: &str = "planet";
/// **Record that a planet is complete** — called by whichever producer just finished one.
///
/// `producer` names the API whose word is being taken (`librqbit …`, `seed symlink`,
/// `declared config.planet`), because a record nobody can attribute is a record nobody
/// can audit.
pub fn record_planet(disk_dir: &Path, planet: &Path, source: &str, producer: &str) -> Result<()> {
let store = state::StateStore::open(disk_dir)?;
let c = state::Completion::of(
state::Step::Planet,
PLANET_KEY,
planet,
source,
producer,
"",
)?;
store.seal(planet, &c)
}
/// **The cached planet the ledger vouches for**, if any.
///
/// # The authority moved (2026-08-05), twice
///
/// First this returned the first entry whose *filename* matched `.osm.bz2` — no
/// completeness check of any kind — so a preallocated, 1.97 %-filled torrent was
/// indistinguishable from a finished planet and got cut into a pyramid stamped fresh.
///
/// The first fix asked the FILESYSTEM (`st_blocks * 512 >= st_size`). Better, and still
/// wrong in kind: it inferred a state librqbit already knew, over a proxy that sparse
/// files, compressing filesystems, preallocation and reflink copies each break on their
/// own.
///
/// Now the producer records completion when its own API says done, and this reads that
/// record ([`planet_standing`]). The filesystem scan below is kept only to CORROBORATE —
/// and, crucially, a candidate the ledger does not vouch for is skipped **however whole
/// it looks**, which is the entire behavioural change.
pub fn cached_planet(dir: &Path) -> Option<PathBuf> {
cached_planet_reporting(dir, &crate::daemon::LogSink::noop())
}
/// [`cached_planet`], but it says out loud what it found and what it REJECTED.
///
/// A skip that leaves no trace is how "the collector sat Offline with a 176 GB planet
/// right there" becomes an unanswerable question. The silent wrapper above exists only
/// for callers that genuinely have no log sink.
pub fn cached_planet_reporting(dir: &Path, log: &crate::daemon::LogSink) -> Option<PathBuf> {
match planet_standing(dir) {
state::Standing::Complete(c) => {
let p = PathBuf::from(&c.artefact);
// Corroboration, not authority: the ledger already said yes. If the second,
// independent signal disagrees, SAY SO and still refuse — two sources that
// disagree are information, and a planet is too expensive to guess about.
if !planet_looks_whole_on_disk(&p) {
log.line(format!(
"planet {} is RECORDED complete by `{}`, but the filesystem's own \
allocation check disagrees. The two independent signals do not \
agree, so this is NOT cut from — investigate rather than re-fetch",
p.display(),
c.producer
));
return None;
}
Some(p)
}
st @ state::Standing::Disagrees { .. } => {
// A FAULT, never a silent re-fetch.
log.line(format!("planet FAULT — {}", st.reason()));
None
}
st => {
// Unrecorded (or foreign). Report what is lying around unvouched-for, so the
// operator can see the difference between "nothing here" and "a file here
// that nobody ever finished".
log.line(format!("no usable planet: {}", st.reason()));
for p in planet_candidates(dir) {
let (on_disk, size) = std::fs::metadata(&p)
.map(|md| (bytes_on_disk(&md), md.len()))
.unwrap_or((0, 0));
#[allow(clippy::float_arithmetic)]
let pct = if size > 0 {
(on_disk as f64) * 100.0 / (size as f64)
} else {
0.0
};
log.line(format!(
" unvouched-for planet candidate {} — {on_disk} of {size} bytes \
allocated ({pct:.2} %). No producer ever recorded it complete, so it \
is not cut from however whole it looks",
p.display()
));
}
None
}
}
}
/// Every file under `dir` shaped like a planet, in a DETERMINISTIC order.
///
/// `read_dir` yields whatever the filesystem feels like, and a collector that cuts from a
/// different planet on two passes over one directory is a collector whose provenance
/// stamp means nothing.
pub fn planet_candidates(dir: &Path) -> Vec<PathBuf> {
let Ok(rd) = std::fs::read_dir(dir) else {
return Vec::new();
};
let mut v: Vec<PathBuf> = rd
.flatten()
.map(|e| e.path())
.filter(|p| {
let n = p
.file_name()
.map(|n| n.to_string_lossy().to_lowercase())
.unwrap_or_default();
n.ends_with(".osm.pbf") || n.ends_with(".osm.bz2") || n.ends_with(".pbf")
})
.collect();
v.sort();
v
}
/// **Does this region's DECLARED extract already hold a complete cut?** — the question
/// that decides whether a pass needs the planet at all ([`OsmCutDaemon::reconcile_once`]
/// step 1) and whether the cut converts or reuses ([`resolve_cut_extract`]). One reader,
/// so those two cannot disagree and leave the pass demanding a planet it never uses.
pub fn declared_extract_complete(config: &serde_json::Value, region: &str) -> bool {
per_region_str(config, "extract_dir", region)
.map(PathBuf::from)
.is_some_and(|d| extract_complete(&d))
}
/// The previously-enumerated region catalog cached on disk, if any and non-empty. Read
/// by [`index_catalog`] (its cache hit) and by a pass that has no planet to scan.
pub fn cached_catalog(disk_dir: &Path) -> Option<Vec<String>> {
std::fs::read(disk_dir.join("catalog.json"))
.ok()
.and_then(|b| serde_json::from_slice::<Vec<String>>(&b).ok())
.filter(|c| !c.is_empty())
}
/// **Enumerate the region catalog from the planet** (design §7) and cache it to
/// `disk_dir/catalog.json`. `osm-katana`'s `stream_admin_region_names`
/// (gatling-parallel, streaming) reads the planet's `boundary=administrative` /
/// `type=boundary|multipolygon` relations and returns each region's `name`
/// (`name:en` fallback) — a BOUNDED-MEMORY pass that builds no in-RAM NodeStore and
/// resolves no geometry (the fix for the ~500 GiB planet-index OOM). A planet with
/// no admin relations (a tiny fixture) yields an empty catalog — a fact, not an
/// error. Reads the cached catalog on a re-run rather than re-scanning.
pub fn index_catalog(planet: &Path, disk_dir: &Path, digest: DigestMode) -> Result<Vec<String>> {
let cache = disk_dir.join("catalog.json");
// **The catalog is keyed on the planet that produced it.**
//
// `catalog.json` IS the region roster — the Settings pick-list, and the list a
// pass with no planet falls back to. Until 2026-08-22 it was written once and
// returned for ever: no planet identity, no mtime, no TTL. A new planet with new
// regions was invisible in the pick-list, and nothing anywhere said the answer was
// frozen — the same defect shape as a server that resolved its map roster at start
// and replayed it (stinky `korpdep-10`, `kvmprov-3`).
//
// The machinery already existed on the two neighbouring axes and this was the one
// that never got it: `stale_planet` guards the EXTRACTS against a planet swap, and
// `planet-identity.json` is correctly keyed on `path|bytes|mtime|mode`. So the fix
// is that same identity, not a new one (LAW 5), written beside the catalog.
let planet_stamp = crate::osm_cut_verify::planet_identity(planet, disk_dir, digest)
.ok()
.map(|id| id.stamp());
let stamp_file = disk_dir.join(CATALOG_PLANET_FILE);
// Cache hit: a previously-enumerated, non-empty catalog on disk, cut from THIS
// planet. An unreadable/absent stamp beside an existing catalog is a catalog from
// before this rule; re-scan once and stamp it, rather than trusting it for ever.
if let Some(cat) = cached_catalog(disk_dir) {
let was = std::fs::read_to_string(&stamp_file)
.ok()
.map(|s| s.trim().to_string());
match (&planet_stamp, &was) {
(Some(now), Some(was)) if now == was => return Ok(cat),
(None, Some(_)) | (None, None) => {
// No identity for this planet (it is not on this box, or stat failed):
// the axis is UNKNOWN, and an unknown must not invalidate a catalog
// that may be the only roster there is. Reuse, and say so.
eprintln!(
"collector: region catalog reused from {} WITHOUT a planet-identity check ({} could not be identified) — the roster may predate a planet swap",
cache.display(),
planet.display()
);
return Ok(cat);
}
(Some(now), was) => eprintln!(
"collector: region catalog at {} was cut from {} and the planet is now {now} — re-scanning rather than serving a frozen pick-list",
cache.display(),
was.as_deref().unwrap_or("an UNRECORDED planet"),
),
}
}
// PLANET-INDEX OOM FIX: the catalog only needs region NAMES, so stream the
// relation blobs for admin/boundary names instead of driving the whole planet
// through `load_admin_boundaries_pbf` (a whole-input in-RAM NodeStore + full
// way/relation materialisation that OOM-killed this host at ~500 GiB). The
// streaming scan builds NO NodeStore and resolves NO geometry — a few MB, not
// hundreds of GiB — and already returns the names sorted + de-duplicated.
let mut names = osm_katana::stream_admin_region_names(&planet.to_string_lossy())
.with_context(|| format!("enumerate admin regions from {}", planet.display()))?;
names.sort();
names.dedup();
// **"No admin relations here" and "I cannot read this format" are DIFFERENT.**
// `stream_admin_region_names` is PBF-only and returns `Ok(vec![])` for a `.bz2`
// planet — the very planet on this box (`planet-260427.osm.bz2`). Reported as a
// fact, that made the daemon reach ONLINE with an empty pick-list AND re-sweep
// 173 GB on every tick, because `cached_catalog` filters empty lists out of the
// cache. An empty result from a format the scanner cannot read is a FAILURE.
if names.is_empty() && !planet_is_pbf(planet) {
anyhow::bail!(
"the region-catalog scan enumerated 0 regions from {} — but that scanner is PBF-only \
and this planet is not a PBF, so the empty result is a FORMAT LIMITATION, not a \
planet without admin relations. Convert the planet to PBF (`osm-katana xml2pbf`) or \
declare the regions directly; an empty pick-list here must never read as a fact",
planet.display()
);
}
// Never write a placeholder catalog: `unwrap_or_default()` on the encode would
// put ZERO bytes at `catalog.json`, and a failed write was invisible — so every
// re-run silently paid for the full planet scan this cache exists to avoid,
// with nothing on disk or on stderr to say why. The names are already in hand,
// so a cache that cannot be written is a warning, not a failure of the call.
match serde_json::to_vec(&names) {
Ok(bytes) => {
if let Err(e) = std::fs::write(&cache, &bytes) {
eprintln!(
"collector: region catalog not cached to {}: {e} — the next run \
will re-scan the planet",
cache.display()
);
} else if let Some(stamp) = &planet_stamp {
// The stamp goes down AFTER the catalog, and only if the catalog
// landed: a stamp beside a catalog that failed to write would claim
// a roster from this planet that nothing produced.
if let Err(e) = std::fs::write(&stamp_file, stamp) {
eprintln!(
"collector: region catalog cached but its planet stamp was not ({}: {e}) \
— the next run re-scans rather than trusting an unkeyed roster",
stamp_file.display()
);
let _ = std::fs::remove_file(&stamp_file);
}
}
}
Err(e) => eprintln!("collector: region catalog not encodable: {e}"),
}
Ok(names)
}
/// **Is this planet a PBF?** — decided on the file's own first bytes, not its name.
///
/// The distinction matters because `osm_katana::stream_admin_region_names` is PBF-only
/// and answers `Ok(vec![])` for anything else. An OSM PBF begins with a big-endian
/// `u32` blob-header length (small — 64 KiB is the format's own cap) followed by the
/// protobuf field for the string `OSMHeader`; a bz2 planet begins `BZh`, an XML one
/// with `<`. Checking the magic rather than the extension means a `.pbf` that is
/// really a bz2 (or the reverse) is still classified correctly.
pub fn planet_is_pbf(planet: &Path) -> bool {
use std::io::Read;
let Ok(mut f) = std::fs::File::open(planet) else {
return false;
};
let mut head = [0u8; 32];
let Ok(n) = f.read(&mut head) else {
return false;
};
if n < 16 {
return false;
}
let len = u32::from_be_bytes([head[0], head[1], head[2], head[3]]) as usize;
if len == 0 || len > 64 * 1024 {
return false;
}
head[4..n].windows(9).any(|w| w == b"OSMHeader")
}
/// A geographic bounding box `(min_lon, min_lat, max_lon, max_lat)` — the
/// memory-safe region clip. Edges inclusive (see [`osm_katana::Bounds`]).
pub type BBox = (f64, f64, f64, f64);
/// **Built-in `region → bbox` table** — the memory-safe clip for the common regions.
/// `(min_lon, min_lat, max_lon, max_lat)`, matched case-insensitively. An unknown
/// region returns `None` (the caller then cuts the whole source, correct only for a
/// tiny fixture / an already-regional extract, or when a config `bbox` override
/// supplies the box). Extend as new regions are needed, or override via config.
fn builtin_region_bbox(region: &str) -> Option<BBox> {
Some(match region.trim().to_ascii_lowercase().as_str() {
// Sweden ≈ lon 10.5..24.2, lat 55.0..69.1 (mainland + Öland/Gotland).
"sweden" => (10.5, 55.0, 24.2, 69.1),
"norway" => (4.0, 57.9, 31.8, 71.4),
"finland" => (19.0, 59.4, 31.6, 70.1),
"denmark" => (7.7, 54.5, 15.3, 57.8),
"iceland" => (-24.6, 63.2, -13.4, 66.6),
"nordics" => (-25.0, 54.0, 42.0, 72.0),
"europe" => (-31.5, 27.0, 69.0, 81.5),
_ => return None,
})
}
/// Parse a config value into a [`BBox`]: a 4-element JSON array
/// `[min_lon, min_lat, max_lon, max_lat]`, or a string parsed by
/// [`osm_katana::Bounds::parse`] (a preset `europe`/`nordics` or a raw
/// `min_lon,min_lat,max_lon,max_lat`). Rejects a bbox whose `max < min`.
fn bbox_from_value(v: &serde_json::Value) -> Option<BBox> {
if let Some(a) = v.as_array() {
if a.len() == 4 {
let f: Vec<f64> = a.iter().filter_map(serde_json::Value::as_f64).collect();
if f.len() == 4 && f[2] >= f[0] && f[3] >= f[1] {
return Some((f[0], f[1], f[2], f[3]));
}
}
return None;
}
if let Some(s) = v.as_str() {
let b = osm_katana::Bounds::parse(s).ok()?;
return Some((b.min_lon, b.min_lat, b.max_lon, b.max_lat));
}
None
}
/// A per-region config `bbox` (or `clip`) override, if any. Accepts either a
/// per-region map `{"sweden": [minlon,minlat,maxlon,maxlat], …}` (looked up by
/// `region`, case-insensitively), a bare 4-array / preset-string applied to every
/// region, so an operator can clip a region the built-in table doesn't know.
fn config_bbox_override(config: &serde_json::Value, region: &str) -> Option<BBox> {
let v = config.get("bbox").or_else(|| config.get("clip"))?;
if let Some(obj) = v.as_object() {
let entry = obj
.get(region)
.or_else(|| obj.get(region.trim().to_ascii_lowercase().as_str()))?;
return bbox_from_value(entry);
}
bbox_from_value(v)
}
/// **A per-region STRING declaration field** — the shape every "one value, possibly
/// per region" config key uses (`poly`, `extract_dir`): either a bare string applied
/// to every region, or a `{"<region>": "<value>"}` map looked up by `region`
/// (case-insensitively). Empty values read as absent. One reader for both keys
/// (LAW 5), so they can never disagree about how a per-region map is spelled.
fn per_region_str(config: &serde_json::Value, key: &str, region: &str) -> Option<String> {
let v = config.get(key)?;
if let Some(obj) = v.as_object() {
let entry = obj
.get(region)
.or_else(|| obj.get(region.trim().to_ascii_lowercase().as_str()))?;
return entry
.as_str()
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string);
}
v.as_str()
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
}
/// **The region's OWN clip polygon, from the declaration** — Osmosis `.poly` /
/// GeoJSON text inline, or a path to a `.poly`/`.geojson` file. Returns the parsed
/// polygon and a human source label for the log.
///
/// This is the SAME polygon data + the SAME parser korp's provisioner clips with
/// (`korp::clip::cut_region` → [`osm_katana::Polygon::from_poly_str`] over
/// `korp/assets/sweden.poly`): the region's shape is a DATUM the caller owns and
/// hands in, not something this collector re-derives. That is why the precise clip
/// costs nothing here and cannot OOM — no planet is read to obtain it.
///
/// Text vs path is decided by shape, not by a flag: a `.poly`/GeoJSON document is
/// multi-line (or starts with `{`), a filesystem path is a single line.
fn resolve_region_poly(
config: &serde_json::Value,
region: &str,
log: &crate::daemon::LogSink,
) -> Option<(osm_katana::Polygon, String)> {
let raw = per_region_str(config, "poly", region)?;
let looks_inline = raw.contains('\n') || raw.starts_with('{');
let (text, source) = if looks_inline {
(
raw.clone(),
format!("the declaration ({} bytes inline)", raw.len()),
)
} else {
match std::fs::read_to_string(&raw) {
Ok(t) => (t, format!("file {raw}")),
Err(e) => {
// A declared polygon that cannot be READ is never silently downgraded
// to the neighbour-admitting bbox without saying so.
log.line(format!(
"region '{region}': declared clip polygon {raw} could not be read ({e}) — \
this cut will NOT use the precise polygon"
));
return None;
}
}
};
let parsed = if text.trim_start().starts_with('{') {
osm_katana::Polygon::from_geojson_str(&text)
} else {
osm_katana::Polygon::from_poly_str(&text)
};
match parsed {
Ok(p) => Some((p, source)),
Err(e) => {
log.line(format!(
"region '{region}': clip polygon from {source} did not parse ({e:#}) — \
this cut will NOT use the precise polygon"
));
None
}
}
}
/// **Resolve the memory-safe bbox clip for a region** (the region-clip fix).
///
/// A from-scratch cut MUST clip to the region's bounding box so [`osm_katana::convert`]
/// STREAMS the planet and keeps only in-bbox features — bounded memory, region-scale
/// output — instead of converting+tiling the WHOLE ~88 GiB planet under a region's
/// name (hours + hundreds of GB + semantically wrong). Resolution order:
/// 1. a config `bbox`/`clip` override (per-region map, a bare 4-array, or a
/// [`osm_katana::Bounds::parse`] string/preset), then
/// 2. the [`builtin_region_bbox`] table (sweden, nordics, europe, …), then
/// 3. `None` — no known bbox ⇒ the caller cuts the whole source (only correct for a
/// tiny fixture / an already-regional extract).
///
/// The axis-aligned bbox is always the memory-safe DEFAULT/FALLBACK. The precise
/// admin-boundary POLYGON refinement is opt-in via `clip_mode=polygon` and layered
/// on top by [`resolve_region_cut_clip`] (which derives the polygon through the
/// bounded streaming subset-scan, NOT the whole-planet in-RAM path that OOMs).
pub fn resolve_region_clip(config: &serde_json::Value, region: &str) -> Option<osm_katana::Bounds> {
let (min_lon, min_lat, max_lon, max_lat) =
config_bbox_override(config, region).or_else(|| builtin_region_bbox(region))?;
Some(osm_katana::Bounds::new(min_lon, min_lat, max_lon, max_lat))
}
/// **Resolve the region's cut clip — the region's OWN POLYGON by DEFAULT, bbox only
/// as the last resort.**
///
/// Resolution, in order:
/// 1. **the declared `.poly`/GeoJSON polygon for this region** ([`resolve_region_poly`])
/// ⇒ [`osm_katana::Clip::Poly`]. This is the default whenever the declaration
/// carries one, needs NO planet scan, and is the exact polygon + parser korp's
/// provisioner clips with.
/// 2. `clip_mode == "polygon"` (opt-in, for a region nobody has a `.poly` for):
/// derive the admin-boundary (multi-)polygon from the planet via the streaming
/// subset-scan [`osm_katana::stream_admin_region_polygon`] (rel → way-ids →
/// node-ids, each pass gatling + hard-capped). Only reachable when step 1 found
/// nothing, because reading a supplied polygon is free and scanning a planet is
/// not — and on a **`.osm.bz2` planet the scan cannot work at all** (it is a
/// PBF-blob scan; see `a_non_pbf_planet_cannot_yield_a_polygon_so_the_poly_is_the_default`).
/// 3. `clip_mode == "bbox"` (explicit) or nothing else resolved: the
/// [`resolve_region_clip`] axis-aligned box as [`osm_katana::Clip::Bbox`].
///
/// # Why the bbox stopped being the default on 2026-08-02
/// The axis-aligned box is memory-safe and **geographically wrong for a country**.
/// Sweden's box (`lon 10.5..24.2, lat 55.0..69.1`) admits parts of Norway, Denmark,
/// Finland, Germany, Poland and the Baltics; every one of those features is real
/// geometry that then owns real tiles under `sweden/tiles/…`. A default collector cut
/// was therefore WORSE than what korp's own one-shot installs, and the precise clip
/// that fixes it was opt-in behind a mode nobody sets. The polygon is now the
/// default and the box is the fallback of last resort, announced as such.
///
/// Returns `None` only when NEITHER a polygon NOR a bbox is known (⇒ the caller
/// cuts the whole source — correct only for a tiny fixture / regional extract).
///
/// Runs the (heavy, 3-pass streaming) planet polygon derivation on the calling
/// thread, so the daemon invokes it from inside `spawn_blocking` — never on the async
/// worker. Step 1 is a file read and is cheap on any thread.
pub fn resolve_region_cut_clip(
config: &serde_json::Value,
region: &str,
planet: &Path,
log: &crate::daemon::LogSink,
) -> Option<osm_katana::Clip> {
let bbox_fallback = |log: &crate::daemon::LogSink| {
let b = resolve_region_clip(config, region);
if let Some(b) = b {
log.line(format!(
"region '{region}': AXIS-ALIGNED BBOX clip lon {:.3}..{:.3} lat {:.3}..{:.3} — \
memory-safe but geographically coarse: it admits every neighbour's geometry \
inside the box. Declare a `.poly` for '{region}' to cut the real shape",
b.min_lon, b.max_lon, b.min_lat, b.max_lat,
));
}
b.map(osm_katana::Clip::Bbox)
};
let mode = config
.get("clip_mode")
.and_then(|v| v.as_str())
.unwrap_or(CLIP_MODE_DEFAULT);
if mode.eq_ignore_ascii_case("bbox") {
// An EXPLICIT `clip_mode=bbox` is the operator's escape hatch and still gets
// the box even when a polygon is at hand — said out loud, because it is a
// deliberate downgrade.
log.line(format!(
"region '{region}': clip_mode=bbox — the precise polygon is SKIPPED by declaration"
));
return bbox_fallback(log);
}
// 1) THE DEFAULT: the region's own polygon, handed in by whoever owns the region
// set. No planet is read, so this can neither be slow nor OOM.
if let Some((poly, source)) = resolve_region_poly(config, region, log) {
let b = poly.bbox();
log.line(format!(
"region '{region}': PRECISE polygon clip from {source} ({} ring(s), {} vertices, \
bbox lon {:.3}..{:.3} lat {:.3}..{:.3}) — no planet scan, no OOM",
poly.ring_count(),
poly.vertex_count(),
b.min_lon,
b.max_lon,
b.min_lat,
b.max_lat,
));
return Some(osm_katana::Clip::Poly(poly));
}
if !mode.eq_ignore_ascii_case("polygon") {
return bbox_fallback(log);
}
match osm_katana::stream_admin_region_polygon(&planet.to_string_lossy(), region) {
Ok(Some(poly)) => {
log.line(format!(
"region '{region}': PRECISE admin-boundary polygon clip \
({} ring(s), {} vertices) — streaming subset-scan, bounded memory",
poly.ring_count(),
poly.vertex_count(),
));
Some(osm_katana::Clip::Poly(poly))
}
Ok(None) => {
log.line(format!(
"region '{region}': clip_mode=polygon but no admin boundary named '{region}' \
in the planet — falling back to the memory-safe bbox clip"
));
bbox_fallback(log)
}
Err(e) => {
log.line(format!(
"region '{region}': polygon clip derivation failed ({e:#}) — falling back to \
the memory-safe bbox clip"
));
bbox_fallback(log)
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The region CUT — its plan (a declaration), its extract reuse, and its stats.
// ─────────────────────────────────────────────────────────────────────────────
/// **Everything one region cut needs, as data** — the plan a caller declares and
/// [`cut_region_extract`] executes. Built from the collector's config by
/// [`RegionCut::from_config`], so the daemon and a test drive the identical path.
pub struct RegionCut<'a> {
/// The planet to cut FROM (only read when no reusable extract is found).
pub planet: &'a Path,
/// The region key — the work-dir name and the log name.
pub region: &'a str,
/// **Where the extract lands** — `<work_root>/<region>`, the very directory
/// korp-server reads the map from. The collector writes INTO the served source;
/// a second location is what let the old pyramid be "complete" and invisible.
pub work_dir: &'a Path,
/// The resolved clip ([`resolve_region_cut_clip`]) — polygon by default.
pub clip: Option<osm_katana::Clip>,
/// **An EXISTING extract to reuse** instead of re-converting the planet
/// (`extract_dir` in the declaration; korp points this at `~/.korp/work/<region>`,
/// the very extract its provisioner cut). Used only when it reads COMPLETE
/// ([`extract_complete`]).
pub extract_dir: Option<PathBuf>,
/// Keep `node_coords.arrow` beside the extract (default TRUE — see
/// [`RegionCut::from_config`]).
pub keep_node_coords: bool,
/// The canonical declaration string ([`cut_declaration`]) recorded in the bucket
/// so a later pass can tell "same declaration ⇒ reuse" from "declaration changed
/// ⇒ regenerate" without a human emptying a prefix.
pub declaration: String,
/// **How many past generations of the extract and the pyramid to KEEP** on disk
/// (config `keep_generations`, default 2). The owner, 2026-08-02: *"historisering
/// med configbar antal, ta bort äldsta vid bortag."* Each swap archives the
/// incumbent as `<name>.gen-NNNN-YYYYMMDD` and prunes the OLDEST beyond this count.
pub keep_generations: usize,
/// How hard to work for the planet's identity — the thing that makes an extract
/// STALE when a new planet exists (config `planet_digest`).
pub planet_digest: crate::osm_cut_verify::DigestMode,
/// **Cooperative stop.** Checked between the stages of a cut, AND bridged onto
/// `osm_katana::CancelToken` for the convert itself (`ConvertOptions.cancel`:
/// a checkpoint per gatling work unit, 1–15 ms stop latency, and a cancelled
/// convert deletes the parquet it opened). A daemon shutdown therefore reaches
/// into a 30–50-minute convert instead of waiting it out.
pub stop: Option<Arc<dyn Fn() -> bool + Send + Sync>>,
/// **What this cut declares must be TRUE of its own output** — the post-conditions
/// [`cut_region_to_store`] runs and FAILS the run on
/// ([`crate::osm_cut_verify::CutExpectations`], config key `expect`). Part of the
/// plan, like everything else here: the declaration says what was asked for, and
/// the checks assert it arrived.
pub expect: crate::osm_cut_verify::CutExpectations,
}
impl<'a> RegionCut<'a> {
/// Has this cut been asked to stop? `false` when no stop seam was wired.
pub fn stopped(&self) -> bool {
self.stop.as_ref().is_some_and(|f| f())
}
/// Build the plan from the collector's declaration (its config JSON).
///
/// `keep_node_coords` defaults to **true**: the collector used to hard-code
/// `false`, which discards `node_coords.arrow` — the side output that gave the
/// decisive "0 out-of-bbox nodes of 96 850 300" proof on 2026-08-02. A cut whose
/// correctness cannot be audited afterwards is a cut nobody can trust, and the
/// file is ~1.5 GB against a 3.5 GB extract, so the default is to keep it.
pub fn from_config(
config: &serde_json::Value,
planet: &'a Path,
region: &'a str,
work_dir: &'a Path,
clip: Option<osm_katana::Clip>,
) -> Self {
Self {
planet,
region,
work_dir,
clip,
extract_dir: per_region_str(config, "extract_dir", region).map(PathBuf::from),
keep_node_coords: config
.get("keep_node_coords")
.and_then(|v| v.as_bool())
.unwrap_or(true),
declaration: cut_declaration(config, region),
keep_generations: config
.get("keep_generations")
.and_then(|v| v.as_u64())
.map(|n| n as usize)
.unwrap_or(2),
planet_digest: DigestMode::from_json(config.get("planet_digest")),
stop: None,
expect: crate::osm_cut_verify::CutExpectations::from_config(config, region),
}
}
}
/// **What one region cut actually did** — the applied output, not a state
/// round-trip: where the extract came from, whether a planet convert ran at all, and
/// every post-condition the cut asserted on the result.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CutStats {
/// True when an existing extract was reused and NO planet convert ran.
pub reused_extract: bool,
/// Where the extract that was cut lives — `<work_root>/<region>`.
pub extract_dir: PathBuf,
/// **Every POST-CONDITION this cut ran, with its numbers and its measured cost.**
/// Carried on the same stats the rest of the run reports through rather than a
/// parallel channel (LAW 5). A cut only returns `Ok` when every one of these
/// passed — a failure is an `Err` that becomes
/// [`crate::daemon::Phase::Failed`] naming the check.
pub checks: Vec<crate::osm_cut_verify::CheckOutcome>,
}
/// Where a convert stages its output before it is published over the live extract —
/// a subdirectory of the work dir, never a sibling (see [`cut_region_extract`]).
pub const STAGING_DIR: &str = ".extract-staging";
/// The same stamp, beside the collector's OWN extract, so a reused extract is only
/// reused for the declaration that produced it.
pub const EXTRACT_STAMP_FILE: &str = "collector-extract.json";
/// Beside that stamp: **WHICH PLANET this extract was cut from, and when that planet
/// was downloaded** ([`crate::osm_cut_verify::PlanetIdentity::stamp`]).
///
/// Kept as its own file rather than folded into the declaration on purpose. The owner,
/// 2026-08-02: an extract is stale when a NEW planet exists to fetch. That is a
/// comparison against the source, and it can only be made when the source is THERE —
/// a declared `extract_dir` must stay tileable on a box with no planet at all, so an
/// unavailable planet must never read as "the source changed".
pub const EXTRACT_PLANET_FILE: &str = "collector-extract-planet.txt";
/// **The planet identity the cached region catalog was enumerated from**, written
/// beside `catalog.json` in the collector's disk dir. Same shape and same producer as
/// [`EXTRACT_PLANET_FILE`] one axis over ([`crate::osm_cut_verify::PlanetIdentity::stamp`]):
/// the catalog is a claim about a planet, so it is keyed on which planet.
pub const CATALOG_PLANET_FILE: &str = "collector-catalog-planet.txt";
/// The default `clip_mode`: try the region's own polygon, else the bbox.
pub const CLIP_MODE_DEFAULT: &str = "auto";
/// A 64-bit FNV-1a digest, hex — a stable, dependency-free identity for the
/// (possibly large) polygon text inside a declaration string. Not a security hash;
/// it only has to change when the polygon changes.
fn digest_hex(bytes: &[u8]) -> String {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for b in bytes {
h ^= *b as u64;
h = h.wrapping_mul(0x100_0000_01b3);
}
format!("{h:016x}")
}
/// **The DECLARATION that produced a pyramid, as one canonical string** — region,
/// zoom ladder, clip mode, the identity of the declared polygon/bbox, and the
/// operator's `revision`. Config-only (no planet, no disk), so the daemon can compare
/// it against the stamp in the bucket BEFORE deciding to spend 35 minutes.
///
/// This is what makes a regeneration expressible in the declaration instead of by
/// hand-emptying an S3 prefix: change any of these (bump `revision`, add a `.poly`,
/// add a zoom) and the next pass sees a different declaration and re-cuts.
///
/// # `exclude_networks` joined it 2026-08-25, and it had to
///
/// The axis is part of the [`osm_katana::tiles::TileRecipe`], so korp's currency judge
/// already answers STALE when it flips — but this daemon's own "same declaration ⇒ reuse"
/// gate is what decides whether anything is re-cut, and it could not see the flip. A
/// pyramid cut with the old value would therefore have been reused forever by the
/// producer while the consumer judged it stale forever: the ladder's failure mode, on a
/// second axis. A declaration that omits a recipe axis is not a declaration of the cut.
///
/// Consequence, stated plainly: the FIRST pass after this change sees a different
/// declaration for every existing collector-cut pyramid and re-cuts it once. That is the
/// intended effect — those pyramids carry the recipe korp calls stale.
pub fn cut_declaration(config: &serde_json::Value, region: &str) -> String {
let mode = config
.get("clip_mode")
.and_then(|v| v.as_str())
.unwrap_or(CLIP_MODE_DEFAULT);
let poly = match per_region_str(config, "poly", region) {
// A path is recorded by NAME + the digest of what it currently holds, so
// editing the file in place is a new declaration too.
Some(raw) => {
let text = if raw.contains('\n') || raw.starts_with('{') {
raw.clone()
} else {
std::fs::read_to_string(&raw).unwrap_or_default()
};
if text.is_empty() {
format!("poly=unreadable({})", digest_hex(raw.as_bytes()))
} else {
format!("poly={}:{}", text.len(), digest_hex(text.as_bytes()))
}
}
None => match resolve_region_clip(config, region) {
Some(b) => format!(
"bbox={:.4},{:.4},{:.4},{:.4}",
b.min_lon, b.min_lat, b.max_lon, b.max_lat
),
None => "clip=none".to_string(),
},
};
let revision = config
.get("revision")
.and_then(|v| v.as_str())
.unwrap_or("");
format!("region={region};clip_mode={mode};{poly};revision={revision}")
}
/// Does this parquet carry its trailing `PAR1` footer magic — i.e. did the writer
/// FINISH it? osm-katana creates every output up front and fills it as passes
/// complete, so the magic is the only cheap on-disk fact separating "this cut
/// finished" from "this cut was interrupted". One `open` + one 4-byte read.
///
/// # The one knowingly-mirrored check, and why (LAW 5)
/// korp's `korp::clip::extract_present_at` asks the identical question about the
/// identical files, and it learned the `.exists()`-is-true-for-a-zero-byte-file lesson
/// the hard way: an interrupted planet clip left `ways.parquet` at 0 bytes and korp
/// would have served Sweden with no roads. The collector must ask the same question
/// before REUSING an extract — but it cannot call korp's copy (a leaf never depends on
/// korp) and korp cannot call this one unconditionally (korp-collectors is an OPTIONAL
/// korp dep, while `korp::clip` compiles in every korp build). So the rule is stated
/// twice on purpose. It is safe to state twice because it is not a policy either side
/// can drift on: it is the parquet format's own trailing `PAR1` footer magic.
pub fn parquet_complete(p: &Path) -> bool {
use std::io::{Read, Seek, SeekFrom};
let Ok(mut f) = std::fs::File::open(p) else {
return false;
};
let Ok(len) = f.metadata().map(|m| m.len()) else {
return false;
};
// 4-byte leading magic + 4-byte footer length + 4-byte trailing magic is the floor
// for any parquet at all; anything shorter cannot be a complete file.
if len < 12 {
return false;
}
if f.seek(SeekFrom::End(-4)).is_err() {
return false;
}
let mut magic = [0u8; 4];
f.read_exact(&mut magic).is_ok() && &magic == b"PAR1"
}
/// **Is there a COMPLETE region extract in `dir`?** Both halves osm-katana writes
/// (`nodes.parquet` + `ways.parquet`) must carry a parquet footer, so a half-written
/// extract reads `false` and is re-cut rather than tiled as if it were data.
pub fn extract_complete(dir: &Path) -> bool {
parquet_complete(&dir.join("nodes.parquet")) && parquet_complete(&dir.join("ways.parquet"))
}
/// **Cut one region's EXTRACT out of the planet (steps b + c).**
///
/// Reuse a finished extract, else convert the planet into `cut.work_dir` — the very
/// directory korp-server reads the map from — with the region's clip. Blocking (a
/// 31–52-minute planet scan); callers run it via `spawn_blocking`.
///
/// `cut.clip` is the region's clip ([`resolve_region_cut_clip`]): when `Some`,
/// `convert` STREAMS the planet and keeps only in-clip nodes/ways (bounded memory,
/// region-scale output); `None` cuts the whole source (only for a tiny fixture / an
/// already-regional extract). The precise [`osm_katana::Clip::Poly`] and the coarse
/// [`osm_katana::Clip::Bbox`] apply through the identical streaming pass-1 per-node
/// test, so neither materialises the planet in RAM.
///
/// # It used to end in a tile pyramid, and no longer does
///
/// Everything after the extract — tile into staging, run the pyramid post-conditions,
/// swap the tree, upload every object, retire the stale ones, write the manifest,
/// stamp the declaration — is **deleted**. korp's map v2 removed the pyramid from the
/// serve path entirely; the artefacts that matter now are the ones the lettered chain
/// produces, and steps d–g of that chain have one implementation, in korp, reached
/// through [`CutDriver`].
///
/// # The safety properties that stayed, because they were never about tiles
/// 1. **Headroom BEFORE the heavy step.** `/home` hit 100 % three times on 2026-08-02,
/// killing a build and a full re-cut. Refusing with two numbers beats discovering it
/// 28 minutes in with a half-written extract on disk.
/// 2. **Never a HALF-SWAPPED extract.** The convert writes into `extract.new`; only
/// once it reads COMPLETE (both parquet footers) is it renamed into place. Until
/// 2026-08-02 the convert wrote STRAIGHT INTO the live extract dir, so an
/// interrupted run left a half-written `nodes.parquet` where a finished one used to
/// be — the only copy, destroyed, by a run that was merely cancelled.
/// 3. **Post-conditions on the extract**, which FAIL the cut rather than publishing a
/// region that admits its neighbours.
pub fn cut_region_extract(cut: &RegionCut<'_>, log: &crate::daemon::LogSink) -> Result<CutStats> {
use crate::osm_cut_verify as verify;
let region = cut.region;
let work = cut.work_dir;
std::fs::create_dir_all(work).with_context(|| format!("mkdir {work:?}"))?;
// **The planet's identity is part of the extract's declaration.** An extract is
// stale when a NEW planet exists to fetch — not when a timer expires — so the
// thing the reuse check compares has to name the source it was built from. A new
// download changes this string and the extract regenerates, by construction.
let planet_id = verify::planet_identity(cut.planet, work, cut.planet_digest).ok();
let planet_stamp = planet_id
.as_ref()
.map(verify::PlanetIdentity::stamp)
.unwrap_or_else(|| "planet=unavailable".to_string());
if let Some(id) = &planet_id {
log.line(format!(
"region '{region}': {} — via {}",
id.stamp(),
id.source
));
}
// NOTE the planet identity is compared SEPARATELY, not folded into this string:
// a declared `extract_dir` must stay reusable with no planet on the box at all
// (that is the whole point of it), and an unavailable planet must therefore never
// read as "the source changed".
let extract_decl = format!(
"{};clip={}",
cut.declaration,
clip_descriptor(cut.clip.as_ref())
);
let now = now_unix();
let mut verdict = verify::CutVerdict::default();
// ── 1) The extract: REUSE a finished one, else convert the planet. ───────────
let (src_dir, reused) = resolve_cut_extract(cut, work, &extract_decl, planet_id.as_ref(), log);
// SAFETY PROPERTY — headroom BEFORE the heavy step.
if !reused {
let head = verify::headroom(
verify::free_bytes(work),
cut.expect.free_bytes_convert,
"the planet convert (a SECOND extract + node_coords + margin)",
);
log.line(format!("region '{region}': post-condition {}", head.line()));
let head_ok = head.ok;
verdict.push(head);
if !head_ok {
anyhow::bail!(
"region '{region}': {}",
verdict
.failure_reason()
.unwrap_or_else(|| "disk headroom".into())
);
}
}
let src_dir = if reused {
src_dir
} else {
// INSIDE `work`, not beside it: `remove_tree_guarded` refuses to delete
// anything outside the directory it is given, and a sibling `work.new` is
// outside. That guard is right — it is what stops a bad path argument from
// taking the planet with it — so the staging dir moves in under its wing.
let staged = work.join(STAGING_DIR);
verify::remove_tree_guarded(work, &staged)?;
std::fs::create_dir_all(&staged).with_context(|| format!("mkdir {staged:?}"))?;
log.line(format!(
"region '{region}': converting the planet → {} (clip {}, keep_node_coords={}) — \
STAGED; the extract now in place is untouched until this one is complete",
staged.display(),
clip_descriptor(cut.clip.as_ref()),
cut.keep_node_coords,
));
// The convert IS cancellable: osm-katana checkpoints per gatling work unit
// (1–15 ms stop latency) and a cancelled run deletes the parquet it opened.
// Bridge the plan's stop seam onto its CancelToken with a 200 ms poller, so
// a daemon shutdown reaches INTO the 30–50-minute convert instead of waiting
// it out between stages.
let cancel = osm_katana::CancelToken::new();
let watcher_done = Arc::new(std::sync::atomic::AtomicBool::new(false));
let watcher = cut.stop.clone().map(|stop| {
let tok = cancel.clone();
let done = Arc::clone(&watcher_done);
std::thread::spawn(move || {
while !done.load(std::sync::atomic::Ordering::SeqCst) {
if stop() {
tok.cancel();
break;
}
std::thread::sleep(std::time::Duration::from_millis(200));
}
})
});
let conv = osm_katana::ConvertOptions {
output_dir: staged.clone(),
clip: cut.clip.clone().map(Arc::new),
// The side output that made the 0-foreign-coordinate audit possible.
keep_node_coords: cut.keep_node_coords,
cancel: Some(cancel),
..Default::default()
};
let converted = osm_katana::convert(cut.planet, &conv)
.with_context(|| format!("convert {:?} → {staged:?}", cut.planet));
watcher_done.store(true, std::sync::atomic::Ordering::SeqCst);
if let Some(h) = watcher {
let _ = h.join();
}
converted?;
anyhow::ensure!(
extract_complete(&staged),
"the convert of region '{region}' finished but {staged:?} holds no COMPLETE extract \
(a nodes/ways parquet without its footer) — REFUSING to swap it over the extract \
now in place"
);
// Stamp the extract with the declaration that produced it, so the NEXT pass
// reuses it only while that declaration still holds. Written INSIDE the staged
// dir, so it becomes visible in the same rename that publishes the extract —
// never a stamp that outlives the thing it describes.
std::fs::write(staged.join(EXTRACT_STAMP_FILE), &extract_decl)
.with_context(|| format!("stamp the staged extract at {staged:?}"))?;
// …and, beside it, WHICH PLANET it was cut from and when that planet was
// downloaded. That is what makes the extract stale when a new planet arrives.
std::fs::write(staged.join(EXTRACT_PLANET_FILE), &planet_stamp)
.with_context(|| format!("record the planet identity at {staged:?}"))?;
// The staged convert is moved OVER the live extract files rather than over the
// whole directory: `work/<region>` also holds `networks/`, the packed table and
// the address index, and renaming the directory would take the whole chain with
// it. Only the files this convert produced are replaced.
publish_staged_extract(&staged, work, cut.keep_generations, now, log)?;
work.to_path_buf()
};
if cut.stopped() {
anyhow::bail!(
"region '{region}': asked to STOP after the extract stage — nothing downstream \
was run, and the chain already on disk is untouched"
);
}
// ── 2) POST-CONDITIONS on the extract. ──────────────────────────────────────
verdict.extend(verify::check_extract(
&src_dir,
cut.clip.as_ref(),
&cut.expect,
log,
));
if let Some(reason) = verdict.failure_reason() {
anyhow::bail!("region '{region}': {reason}");
}
// ── THE COMPLETION MOMENTS for b and c ──────────────────────────────────────
record_chain_steps(cut, &src_dir, &planet_stamp, log)
.context("sealing the chain's completion records")?;
Ok(CutStats {
reused_extract: reused,
extract_dir: src_dir,
checks: verdict.checks,
})
}
/// **Move a staged convert's files over the live extract, keeping the rest of the
/// chain in place.**
///
/// The old code renamed the whole extract DIRECTORY. That was correct when the
/// directory held nothing but the extract; `work/<region>` now also holds `networks/`,
/// the packed base table and `address.parquet`, and a directory swap would silently
/// take the whole chain with it — a "successful" convert that deleted steps e, f and
/// g. So the swap is per-FILE, and each replaced file is archived as a generation
/// exactly the way [`crate::osm_cut_verify::swap_into_place_gen`] archives a directory.
fn publish_staged_extract(
staged: &Path,
work: &Path,
keep_generations: usize,
now: u64,
log: &crate::daemon::LogSink,
) -> Result<()> {
use crate::osm_cut_verify as verify;
// The entries are COLLECTED before the first rename. Renaming files out of a
// directory while its `ReadDir` is still open is exactly the shape that made half
// the swap find nothing left to move.
let entries: Vec<PathBuf> = std::fs::read_dir(staged)
.with_context(|| format!("read {staged:?}"))?
.filter_map(|e| e.ok().map(|e| e.path()))
.collect();
anyhow::ensure!(
!entries.is_empty(),
"the convert staged NOTHING at {staged:?} — refusing to report a published extract \
that does not exist"
);
let mut moved = Vec::new();
for from in entries {
let Some(name) = from.file_name().map(|n| n.to_os_string()) else {
continue;
};
let to = work.join(&name);
// Archive the incumbent as a numbered, dated generation rather than clobbering
// it: `keep_generations` is the owner's "historisering med configbar antal".
if to.exists() {
let n = verify::next_generation(&to);
let slot = verify::generation_path(&to, n, now);
verify::remove_tree_guarded(work, &slot)?;
std::fs::rename(&to, &slot)
.with_context(|| format!("archive the incumbent {to:?} as {slot:?}"))?;
}
std::fs::rename(&from, &to).with_context(|| format!("swap {from:?} into {to:?}"))?;
verify::prune_generations(work, &to, keep_generations);
moved.push(name.to_string_lossy().to_string());
}
verify::remove_tree_guarded(work, staged)?;
log.line(format!(
"published the staged convert into {} ({})",
work.display(),
moved.join(", ")
));
Ok(())
}
/// **Seal b and c** — one completion record per step this collector produced, each
/// naming the artefact it finished, the source it came from and the declaration in
/// force.
///
/// Both go through [`state::StateStore::seal`], which fsyncs the artefact before
/// committing the row: a record therefore *implies* the artefact was finished, and a
/// crash in between leaves no record rather than a durable claim about a file that never
/// landed.
///
/// **Steps d–g are sealed by their own producer, not here.** They are korp's chain
/// (`regions::start_or_resume`), reached through [`CutDriver`], and a record written on
/// their behalf by a process that did not produce them would be exactly the "bad
/// assertion" this store exists to replace.
///
/// A step whose artefact is absent is simply not recorded — this run did not produce it.
fn record_chain_steps(
cut: &RegionCut<'_>,
src_dir: &Path,
planet_stamp: &str,
log: &crate::daemon::LogSink,
) -> Result<()> {
let store = state::StateStore::open(cut.work_dir)?;
let decl = &cut.declaration;
let region = cut.region;
// b, c — the extract's DEFINING artefact per step. One record per step, naming the
// file whose completion IS that step's completion; the step's other outputs
// (`nodes.parquet` beside the coords, `relations.parquet` beside the ways) are
// produced by the same `finalize()` and cannot land without it.
let steps: [(state::Step, &str); 2] = [
(state::Step::Coords, "node_coords.arrow"),
(state::Step::Tables, "ways.parquet"),
];
for (step, file) in steps {
let p = src_dir.join(file);
if !p.is_file() {
continue;
}
let c = state::Completion::of(
step,
region,
&p,
planet_stamp,
"osm-katana convert (finalize closed a complete file)",
decl,
)?;
store.seal(&p, &c)?;
log.line(format!(
"region '{region}': RECORDED complete — {}",
c.line()
));
}
Ok(())
}
/// A short, stable description of a clip — what a stamp records and a log prints.
fn clip_descriptor(clip: Option<&osm_katana::Clip>) -> String {
match clip {
None => "none(whole source)".to_string(),
Some(osm_katana::Clip::Bbox(b)) => format!(
"bbox({:.4},{:.4},{:.4},{:.4})",
b.min_lon, b.min_lat, b.max_lon, b.max_lat
),
Some(osm_katana::Clip::Poly(p)) => {
let b = p.bbox();
format!(
"poly({} ring(s),{} vertices,bbox {:.4},{:.4},{:.4},{:.4})",
p.ring_count(),
p.vertex_count(),
b.min_lon,
b.min_lat,
b.max_lon,
b.max_lat
)
}
}
}
/// **Which extract this cut will tile, and whether the planet convert is skipped.**
///
/// Candidates, in order:
/// 1. the declared `extract_dir` (korp points it at `~/.korp/work/<region>`, the
/// extract its own provisioner cut) — reused when it reads COMPLETE **and** does
/// not carry a stamp naming a DIFFERENT declaration. An unstamped one is still
/// reused (korp does not stamp what it cuts) but is announced as unverified: the
/// extract post-conditions are then the only thing standing between a stale
/// extract and a chain that is declared against the new polygon and built from
/// the old data — the "green and wrong at the same time" case.
/// 2. the extract already in `work/<region>` from a previous pass — reused only when
/// it is complete AND its [`EXTRACT_STAMP_FILE`] matches THIS declaration, so a
/// changed clip is never cut from a stale extract.
/// 3. otherwise: convert (into `<work>.new`, swapped into place once complete — see
/// [`cut_region_extract`]).
///
/// Returns `(dir_to_tile, reused)`; when `reused` is false the caller stages the
/// convert rather than writing over `own_extract`, so a cancelled run never destroys
/// the extract that is there.
fn resolve_cut_extract(
cut: &RegionCut<'_>,
own_dir: &Path,
extract_decl: &str,
planet: Option<&crate::osm_cut_verify::PlanetIdentity>,
log: &crate::daemon::LogSink,
) -> (PathBuf, bool) {
let region = cut.region;
// **Is this extract built from a planet that has since been replaced?** Only
// answerable when the planet is actually present; when it is not, the axis is
// unknown and the extract is judged on its declaration alone.
let stale_planet = |dir: &Path| -> Option<String> {
let now = planet?.stamp();
let was = std::fs::read_to_string(dir.join(EXTRACT_PLANET_FILE)).ok()?;
let was = was.trim().to_string();
(!was.is_empty() && was != now)
.then(|| format!("was cut from {was}, the planet is now {now}"))
};
if let Some(declared) = &cut.extract_dir {
if extract_complete(declared) {
// **A borrowed extract is not exempt from the declaration.** Until
// 2026-08-02 this branch reused a declared `extract_dir` on the strength of
// two parquet footers ALONE — no stamp compared, ever. The consequence,
// measured that day: changing the clip polygon changes the DECLARATION, so
// the chain is re-declared against the NEW polygon — but it is built from
// the extract the OLD polygon produced. Green and wrong at the same time.
let stamped =
std::fs::read_to_string(declared.join(EXTRACT_STAMP_FILE)).unwrap_or_default();
if let Some(why) = stale_planet(declared) {
log.line(format!(
"region '{region}': declared extract_dir {} is STALE — {why}; a new planet exists, so the extract is re-cut",
declared.display()
));
} else if !stamped.is_empty() && stamped != extract_decl {
log.line(format!(
"region '{region}': declared extract_dir {} was cut for a DIFFERENT \
declaration ({stamped}) than this pass runs with ({extract_decl}) — \
converting the planet instead of reusing a stale extract",
declared.display()
));
} else {
log.line(format!(
"region '{region}': REUSING the declared extract at {} (nodes+ways parquet \
both carry their footer{}) — skipping the planet convert",
declared.display(),
if stamped.is_empty() {
"; UNSTAMPED, so its provenance is unverified and the extract \
post-conditions are the only thing checking it against this declaration"
} else {
"; stamped with THIS declaration"
},
));
return (declared.clone(), true);
}
} else {
log.line(format!(
"region '{region}': declared extract_dir {} holds no COMPLETE extract (a missing \
or half-written nodes/ways parquet) — converting the planet instead",
declared.display()
));
}
}
if extract_complete(own_dir) {
let stamped = std::fs::read_to_string(own_dir.join(EXTRACT_STAMP_FILE)).unwrap_or_default();
if let Some(why) = stale_planet(own_dir) {
log.line(format!(
"region '{region}': own extract at {} is STALE — {why}; re-converting",
own_dir.display()
));
} else if stamped == extract_decl {
log.line(format!(
"region '{region}': REUSING this collector's own extract at {} (same declaration) \
— skipping the planet convert",
own_dir.display()
));
return (own_dir.to_path_buf(), true);
} else {
log.line(format!(
"region '{region}': own extract at {} was cut for a DIFFERENT declaration \
({}) — re-converting",
own_dir.display(),
if stamped.is_empty() {
"unstamped".to_string()
} else {
stamped
},
));
}
}
(own_dir.to_path_buf(), false)
}
fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::daemon::{
BiosLog, Fetcher, LogSink, MemObjectStore, MemWarehouse, PhaseSink, Shutdown,
};
const FIXTURE_PBF: &[u8] = include_bytes!("fixtures/testland-latest.osm.pbf");
fn tmp_dir(tag: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("osm_cut_{tag}_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
std::fs::create_dir_all(&d).unwrap();
d
}
fn seed_fixture(dir: &Path) -> PathBuf {
let p = dir.join("testland-latest.osm.pbf");
std::fs::write(&p, FIXTURE_PBF).unwrap();
p
}
#[test]
fn config_defaults_and_schema_pick_list() {
let d = OsmCutDaemon::new();
assert_eq!(d.name(), NAME);
assert!(!d.enabled(), "off until enabled in Settings");
assert_eq!(d.regions(), vec!["sweden".to_string()]);
// DERIVED, not spelled — for the same reason `default_through_step` is.
//
// b1716fc made `through_step()` return the chain's last step and updated the two
// assertions in `tests/osm_cut_is_the_pipeline.rs`. This one is a THIRD copy of
// the same pin, in `--lib` rather than that integration file, and it was left
// reading `Step::Address` — so `cargo test --features osm-daemon --lib` has been
// red on main since, with `left: Buildings, right: Address`. The commit verified
// with `--test osm_cut_is_the_pipeline` alone, which is exactly the file this
// copy is NOT in.
assert_eq!(
d.through_step(),
default_through_step(),
"the default cut runs the WHOLE chain, ending at its last step"
);
assert_eq!(d.driver_name(), "none (leaf-only build)");
// The schema carries a region pick-list (empty before the planet is indexed).
let schema = d.schema();
let region_field = schema.fields.iter().find(|f| f.key == "regions").unwrap();
assert!(
matches!(®ion_field.kind, FieldKind::RegionPickList { catalog } if catalog.is_empty())
);
}
/// ★ **A planet is usable because it was RECORDED complete — never because a file of
/// that name is lying there.**
///
/// This test used to read `seed_fixture(&dir); assert!(cached_planet(&dir).is_some())`
/// — it asserted the old contract, in which the presence of a `.osm.pbf` WAS the
/// answer, and it passed happily over the defect that shipped: a preallocated,
/// 1.97 %-filled torrent satisfied it exactly as well as a finished planet.
///
/// The middle assertion is the whole change. A perfectly whole, fully-allocated file
/// that no producer ever vouched for reads `None`, and the ONLY thing that moves it
/// to `Some` is a completion record.
#[test]
fn a_planet_is_usable_only_once_a_producer_has_recorded_it_complete() {
let dir = tmp_dir("cache");
assert!(cached_planet(&dir).is_none(), "empty cache ⇒ none");
let planet = seed_fixture(&dir);
assert!(planet.is_file() && std::fs::metadata(&planet).unwrap().len() > 0);
assert!(
planet_looks_whole_on_disk(&planet),
"the fixture must LOOK whole to the filesystem, or the next assertion proves \
nothing about where the authority lives"
);
// ★ RED under the old contract: a whole file that nobody finished is NOT usable.
assert!(
cached_planet(&dir).is_none(),
"an unrecorded planet was handed out. The filesystem is deciding again, which \
is exactly the defect: it cannot tell a finished download from a preallocated \
one, and librqbit could have told us"
);
assert_eq!(planet_standing(&dir).kind(), "unrecorded");
// The producer says done → and only now is it usable.
record_planet(&dir, &planet, "test seed", "test producer").unwrap();
assert_eq!(planet_standing(&dir).kind(), "complete");
assert_eq!(
cached_planet(&dir).as_deref(),
Some(planet.as_path()),
"a RECORDED planet must be handed back"
);
}
/// ★ **Record present, artefact truncated ⇒ the two signals DISAGREE, and that is
/// reported as a fault — not silently re-fetched and not silently served.**
///
/// This is the corroboration leg: the ledger alone would say "complete", the
/// filesystem alone would say "fine, it's fully allocated". Only holding both
/// catches a sealed artefact that something else damaged.
#[test]
fn a_recorded_planet_that_shrank_is_a_reported_fault_not_a_silent_refetch() {
let dir = tmp_dir("fault");
let planet = seed_fixture(&dir);
record_planet(&dir, &planet, "test seed", "test producer").unwrap();
assert_eq!(planet_standing(&dir).kind(), "complete");
let full = std::fs::metadata(&planet).unwrap().len();
std::fs::OpenOptions::new()
.write(true)
.open(&planet)
.unwrap()
.set_len(full / 2)
.unwrap();
let st = planet_standing(&dir);
assert_eq!(st.kind(), "disagrees", "{}", st.reason());
assert!(
st.is_fault(),
"a damaged sealed artefact must surface as a FAULT"
);
assert!(st.reason().contains("SHRUNK"), "{}", st.reason());
assert!(
cached_planet(&dir).is_none(),
"a planet the ledger and the disk disagree about must not be cut from"
);
// And the log says so, rather than the collector going quiet.
let (log, lines) = crate::daemon::LogSink::recorder();
let _ = cached_planet_reporting(&dir, &log);
let said = lines.lock().unwrap().join("\n");
assert!(
said.contains("FAULT"),
"the fault must be REPORTED, not swallowed: {said}"
);
}
/// **Region catalog enumeration from a real planet fixture** (design §7). The
/// tiny testland fixture has no admin relations → an empty catalog (a FACT, not
/// an error), and the result is cached to `catalog.json`.
#[test]
fn index_catalog_from_fixture_is_a_fact_and_cached() {
let dir = tmp_dir("catalog");
let planet = seed_fixture(&dir);
let cat = index_catalog(&planet, &dir, DigestMode::Sidecar)
.expect("catalog enumerates (real OSM bytes)");
// The 4-node/2-way fixture carries no boundary relations → empty pick-list.
assert!(
cat.is_empty(),
"no admin regions in the tiny fixture (a fact)"
);
assert!(dir.join("catalog.json").is_file(), "catalog cached to disk");
}
/// **PLANET-INDEX OOM FIX (bounded-memory catalog).** `index_catalog` over a
/// SMALL MULTI-region synthetic PBF enumerates the admin-region NAMES via the
/// streaming relation scan — the path that replaced the whole-planet in-RAM
/// `load_admin_boundaries_pbf` (507 GiB OOM). We build the PBF from a synthetic
/// `.osm` through osm-katana's own public XML→PBF converter (no committed binary
/// fixture, no private encoder), then assert the catalog is exactly the two
/// ADMINISTRATIVE regions — enumerated WITHOUT building any in-RAM NodeStore (the
/// `index_catalog` → `stream_admin_region_names` code path carries none) — and is
/// cached to `catalog.json`. A NAMED `type=multipolygon` building (the 1.34 M
/// planet-junk bug) and a `type=route` relation are both excluded.
#[test]
fn index_catalog_enumerates_multiregion_names_without_node_store() {
let dir = tmp_dir("multiregion");
// Two admin/boundary regions + one non-boundary route relation to prove the
// filter. Nodes/way exist only so the relations are well-formed OSM.
// Two real admin regions (boundary=administrative + admin_level + name);
// one NAMED type=multipolygon building that MUST be excluded (the 1.34 M
// planet-junk bug); one type=route that MUST be excluded.
let osm = r#"<?xml version="1.0" encoding="UTF-8"?>
<osm version="0.6" generator="korp-collectors-test">
<node id="1" lat="0.0" lon="0.0"/>
<node id="2" lat="0.0" lon="1.0"/>
<node id="3" lat="1.0" lon="1.0"/>
<node id="4" lat="1.0" lon="0.0"/>
<way id="10">
<nd ref="1"/><nd ref="2"/><nd ref="3"/><nd ref="4"/><nd ref="1"/>
</way>
<relation id="100">
<member type="way" ref="10" role="outer"/>
<tag k="type" v="boundary"/>
<tag k="boundary" v="administrative"/>
<tag k="admin_level" v="4"/>
<tag k="name" v="Alpha"/>
</relation>
<relation id="150">
<member type="way" ref="10" role="outer"/>
<tag k="type" v="boundary"/>
<tag k="boundary" v="administrative"/>
<tag k="admin_level" v="6"/>
<tag k="name" v="Delta"/>
</relation>
<relation id="200">
<member type="way" ref="10" role="outer"/>
<tag k="type" v="multipolygon"/>
<tag k="building" v="yes"/>
<tag k="name" v="Beta Tower"/>
</relation>
<relation id="300">
<member type="way" ref="10" role="forward"/>
<tag k="type" v="route"/>
<tag k="route" v="bus"/>
<tag k="name" v="Route 66"/>
</relation>
</osm>
"#;
let osm_path = dir.join("multiregion.osm");
std::fs::write(&osm_path, osm).unwrap();
let pbf_path = dir.join("multiregion.osm.pbf");
osm_katana::xml_to_pbf::xml_to_pbf_raw_path(&osm_path, &pbf_path, 1)
.expect("synthetic .osm → PBF via osm-katana");
let cat = index_catalog(&pbf_path, &dir, DigestMode::Sidecar)
.expect("catalog enumerates from the PBF");
assert_eq!(
cat,
vec!["Alpha".to_string(), "Delta".to_string()],
"streaming index enumerates the two ADMINISTRATIVE regions and excludes \
the named type=multipolygon building and the route relation: {cat:?}",
);
assert!(dir.join("catalog.json").is_file(), "catalog cached to disk");
let _ = std::fs::remove_dir_all(&dir);
}
/// **RED: a NEW planet must not be answered with the OLD planet's region roster.**
///
/// `catalog.json` is the region pick-list. It used to be written once and returned
/// for ever — no planet identity, no mtime, no TTL — so a planet swap that adds or
/// renames regions was invisible in Settings, and nothing said the answer was
/// frozen (stinky `korpdep-10`). The two neighbouring axes already had the cure:
/// the EXTRACTS are guarded against a planet swap by `stale_planet`, and
/// `planet-identity.json` is keyed on `path|bytes|mtime|mode`. The catalog was the
/// only one of the three that never got it.
///
/// Two synthetic planets in the SAME disk dir, one region each, different names.
/// Before the fix the second scan returned `["Alpha"]`.
#[test]
fn a_new_planet_is_not_answered_with_the_old_planets_catalog() {
let dir = tmp_dir("catalog_identity");
let pbf_for = |name: &str, file: &str| -> PathBuf {
let osm = format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
<osm version="0.6" generator="korp-collectors-test">
<node id="1" lat="0.0" lon="0.0"/>
<node id="2" lat="0.0" lon="1.0"/>
<node id="3" lat="1.0" lon="1.0"/>
<node id="4" lat="1.0" lon="0.0"/>
<way id="10">
<nd ref="1"/><nd ref="2"/><nd ref="3"/><nd ref="4"/><nd ref="1"/>
</way>
<relation id="100">
<member type="way" ref="10" role="outer"/>
<tag k="type" v="boundary"/>
<tag k="boundary" v="administrative"/>
<tag k="admin_level" v="4"/>
<tag k="name" v="{name}"/>
</relation>
</osm>
"#
);
let x = dir.join(format!("{file}.osm"));
std::fs::write(&x, osm).unwrap();
let pbf = dir.join(format!("{file}.osm.pbf"));
osm_katana::xml_to_pbf::xml_to_pbf_raw_path(&x, &pbf, 1).expect("xml -> pbf");
pbf
};
let planet_a = pbf_for("Alpha", "planet-a");
let first = index_catalog(&planet_a, &dir, DigestMode::Sidecar).expect("first scan");
assert_eq!(
first,
vec!["Alpha".to_string()],
"the first planet's roster"
);
let stamped = std::fs::read_to_string(dir.join(CATALOG_PLANET_FILE))
.expect("the catalog records WHICH planet it came from");
assert!(
stamped.starts_with("planet="),
"the stamp is the same PlanetIdentity form the extracts use: {stamped}"
);
// A cache HIT on the same planet must not re-scan — the whole point of the
// file. Proven by deleting the planet: a scan would fail, a hit cannot.
let hit = index_catalog(&planet_a, &dir, DigestMode::Sidecar).expect("cache hit");
assert_eq!(hit, first, "same planet ⇒ the cached roster, no re-scan");
// A DIFFERENT planet in the same dir. Before the fix this returned ["Alpha"].
let planet_b = pbf_for("Beta", "planet-b");
let second = index_catalog(&planet_b, &dir, DigestMode::Sidecar).expect("second scan");
assert_eq!(
second,
vec!["Beta".to_string()],
"a new planet must be RE-SCANNED, not answered with the old roster",
);
let restamped = std::fs::read_to_string(dir.join(CATALOG_PLANET_FILE)).unwrap();
assert_ne!(
restamped, stamped,
"…and the recorded identity moves with it, or the next run freezes again",
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **The daemon cuts a region's chain and REPORTS every step by letter.**
///
/// Seed the fixture as the planet, drive one reconcile pass with the in-memory
/// seams through step `c`, and assert the applied output: the extract on disk, a
/// step report per letter, a summary row landed per step, the phase sequence
/// naming the letter it was on, and ONLINE on the BIOS bus.
///
/// It stops at `c` because steps d–g are korp's ([`CutDriver`]) and this build
/// injects none — the honest end of a leaf-only cut. The `Unavailable` half is
/// proven separately by [`a_cut_without_a_driver_is_incomplete_and_says_which_step`].
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_daemon_cuts_a_region_and_reports_every_step_by_letter() {
let dir = tmp_dir("cut_steps");
let planet = seed_fixture(&dir);
let work = dir.join("work");
let warehouse = Arc::new(MemWarehouse::new());
let bios = BiosLog::new();
let (phase_sink, phases) = PhaseSink::recorder();
let cfg = serde_json::json!({
"regions": ["testland"],
"planet": planet.to_string_lossy(),
"work_dir": work.to_string_lossy(),
"through_step": "c",
// A CLIP and a presence claim, because a cut is a geographic verdict:
// without them this pass asserts nothing about the map.
"bbox": [10.5, 55.0, 24.2, 69.1],
"expect": { "present": [
{ "name": "Testland", "lon": 17.93, "lat": 59.36, "radius_km": 60 }
]},
});
let ctx = CollectorCtx {
name: NAME.into(),
disk_dir: dir.clone(),
config: cfg.clone(),
bucket: Arc::new(MemObjectStore::new()),
warehouse: warehouse.clone(),
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: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
let daemon = OsmCutDaemon::with_config(true, cfg);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_eq!(
daemon.phase(),
Phase::Online,
"a complete a–c chain is ONLINE: {:?}",
daemon.snapshot().error
);
assert!(daemon.healthy());
// THE APPLIED OUTPUT: the extract is on disk, in the work dir korp reads.
let region_dir = work.join("testland");
assert!(
parquet_complete(®ion_dir.join("nodes.parquet")),
"step b produced a complete nodes.parquet in {}",
region_dir.display()
);
assert!(
parquet_complete(®ion_dir.join("ways.parquet")),
"step c produced a complete ways.parquet"
);
// One report per LETTER, in chain order, all complete.
let snap = daemon.snapshot();
let report = snap.reports.first().expect("a report for testland");
assert_eq!(report.region, "testland");
assert_eq!(
report.steps.iter().map(|s| s.letter).collect::<Vec<_>>(),
vec!['a', 'b', 'c'],
"the pass reports the pipeline's own letters"
);
assert!(report.complete(), "{}", report.verdict());
let b = &report.steps[1];
assert_eq!(b.step, "coords");
assert_eq!(b.state, StepState::Produced, "the first pass CONVERTS");
assert!(b.rows > 0, "step b measured real rows: {b:?}");
assert!(b.bytes > 0, "step b measured real bytes: {b:?}");
// It lands NO map RecordBatch — one SUMMARY row per step, and nothing else.
assert_eq!(
crate::daemon::WarehouseSink::rows(warehouse.as_ref(), SUMMARY_TABLE),
3,
"one summary row per reported step"
);
// The phase sink saw the state machine, naming the letter it was on.
let seq = phases.lock().unwrap().clone();
assert!(
seq.iter().any(|p| matches!(p, Phase::Cutting { region, .. } if region.contains("b coords"))),
"a phase named step b by letter: {seq:?}"
);
assert_eq!(seq.last(), Some(&Phase::Online));
assert!(
bios.is_online(NAME),
"collector tuned in to ONLINE on the BIOS bus"
);
// Restart-safe: a second pass REUSES the extract and converts nothing.
let (phase2, _) = PhaseSink::recorder();
let cfg2 = daemon.snapshot();
let _ = cfg2;
let ctx2 = CollectorCtx {
name: NAME.into(),
disk_dir: dir.clone(),
config: serde_json::json!({}),
bucket: Arc::new(MemObjectStore::new()),
warehouse: warehouse.clone(),
fetch: Fetcher::noop(),
phase: phase2,
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: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
daemon.reconcile_once(&ctx2).await.expect("reconcile #2");
let again = daemon.snapshot();
let r2 = again.reports.first().expect("a second report");
assert!(
r2.steps
.iter()
.all(|s| s.state == StepState::AlreadyComplete),
"restart-safe: NOTHING was re-cut — {}",
r2.verdict()
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **A cut with no [`CutDriver`] is INCOMPLETE, and says at which step.**
///
/// Steps d–g live in korp. A build that cannot run them must report `unavailable`
/// by name and land DEGRADED — never a skipped step that reads as done. This is the
/// gate on the honest half of the seam: the collector's verdict is about the whole
/// chain, not about the part it happens to be able to do.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_cut_without_a_driver_is_incomplete_and_says_which_step() {
let dir = tmp_dir("no_driver");
let planet = seed_fixture(&dir);
let work = dir.join("work");
let cfg = serde_json::json!({
"regions": ["testland"],
"planet": planet.to_string_lossy(),
"work_dir": work.to_string_lossy(),
"bbox": [10.5, 55.0, 24.2, 69.1],
});
let ctx = CollectorCtx {
name: NAME.into(),
disk_dir: dir.clone(),
config: cfg.clone(),
bucket: Arc::new(MemObjectStore::new()),
warehouse: Arc::new(MemWarehouse::new()),
fetch: Fetcher::noop(),
phase: PhaseSink::noop(),
log: LogSink::noop(),
bios: BiosLog::new().bus(),
shutdown: Shutdown::new(),
secret: std::sync::Arc::new(crate::daemon::EnvSecretStore::new()),
marks: std::sync::Arc::new(crate::freshness::MemMarkStore::new()),
nudge: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
let daemon = OsmCutDaemon::with_config(true, cfg);
daemon.reconcile_once(&ctx).await.expect("the pass runs");
let snap = daemon.snapshot();
let report = snap.reports.first().expect("a report");
assert!(!report.complete(), "a chain missing d–g is NOT complete");
let gap = report.first_gap().expect("a named gap");
assert_eq!(gap.letter, 'd', "the first gap is step d: {}", gap.line());
assert_eq!(gap.state, StepState::Unavailable);
assert!(
gap.note.contains("korp") && gap.note.contains("CutDriver"),
"the note names WHO owns the step and WHY it did not run: {}",
gap.note
);
match daemon.phase() {
Phase::Degraded { reason } => assert!(
reason.contains("incomplete at d"),
"the phase carries the gap: {reason}"
),
other => panic!("an incomplete chain must land DEGRADED, got {other:?}"),
}
let _ = std::fs::remove_dir_all(&dir);
}
/// **A catalog SCAN FAILURE must not be laundered into an empty catalog.**
///
/// `index_catalog` returning `Err` used to be `unwrap_or_default()`ed, so a
/// broken planet scan produced `[]` — which is INDISTINGUISHABLE from the
/// documented legitimate case (a planet with no admin relations yields an empty
/// catalog, "a fact, not an error"). Both logged "region catalog: 0 region(s)"
/// and both went on to report ONLINE. RAGNARÖK: landing nothing is RED unless
/// the reason says why.
///
/// Injected as a re-run over an UNREADABLE planet whose region tiles are already
/// in the bucket, so the restart-safe path skips the cut and the enumeration is
/// the only thing that fails. The pass must end DEGRADED carrying the reason.
///
/// The injection is a mode-000 planet (the real shape: a 170 GB file on a mount
/// that lost permissions) rather than a corrupt one, because a corrupt PBF does
/// NOT fail — `osm_katana::stream_admin_region_names` only errors on open/mmap,
/// so garbage bytes scan to zero blobs and return `Ok([])`. Worth knowing: that
/// is a second, separate silent-empty path living in osm-katana.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_failed_catalog_scan_lands_degraded_not_online() {
use std::os::unix::fs::PermissionsExt;
let dir = tmp_dir("catalog_fail");
// Exists (so `resolve_planet` accepts it) but cannot be opened.
let planet = dir.join("unreadable-latest.osm.pbf");
std::fs::write(&planet, b"planet bytes").unwrap();
std::fs::set_permissions(&planet, std::fs::Permissions::from_mode(0o000)).unwrap();
if std::fs::File::open(&planet).is_ok() {
// Running as root: the mode cannot deny us, so the failure cannot be
// injected. Skip rather than pass vacuously.
let _ = std::fs::remove_dir_all(&dir);
return;
}
// The region's extract is already COMPLETE, so steps b and c reuse it and the
// catalog failure is the ONLY fault in this pass. (Before the pyramid was
// deleted this was "the tiles are already in the bucket"; same idea, on the
// artefact that replaced them.)
let cfg = daemon_config_for(&planet);
let work = PathBuf::from(cfg["work_dir"].as_str().unwrap()).join("testland");
std::fs::create_dir_all(&work).unwrap();
for f in ["nodes.parquet", "ways.parquet"] {
std::fs::write(work.join(f), b"PAR1\x00\x00\x00\x00PAR1").unwrap();
}
assert!(extract_complete(&work), "the pass has nothing to convert");
let bios = BiosLog::new();
let (phase_sink, phases) = PhaseSink::recorder();
let ctx = CollectorCtx {
name: NAME.into(),
disk_dir: dir.clone(),
config: cfg.clone(),
bucket: Arc::new(MemObjectStore::new()),
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: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
let daemon = OsmCutDaemon::with_config(true, cfg);
daemon
.reconcile_once(&ctx)
.await
.expect("the pass completes — degraded is a fact, not a hard error");
match daemon.phase() {
Phase::Degraded { reason } => assert!(
reason.contains("region catalog"),
"the reason must name the catalog: {reason}"
),
other => panic!("a failed catalog scan must land DEGRADED, got {other:?}"),
}
// The run state carries it too, not just the transient phase.
assert!(
daemon.snapshot().error.is_some(),
"run state records the reason"
);
let seq = phases.lock().unwrap().clone();
assert!(
matches!(seq.last(), Some(Phase::Degraded { .. })),
"the pass must not end green with an unknown catalog: {seq:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// A daemon declaration for the fixture, **always cutting into a dir of its own**.
///
/// `work_dir` is not optional in a test. Left blank it resolves to
/// [`default_work_dir`] — `~/.korp/work` — which is the LIVE cut korp-server is
/// serving from, and a test that converts a fixture planet into it would overwrite
/// a region the map is using. The declaration is the only place that can be said,
/// so it is said here, once, for every test that builds one.
fn daemon_config_for(planet: &Path) -> serde_json::Value {
let work = planet
.parent()
.unwrap_or(Path::new("."))
.join("work-of-this-test");
serde_json::json!({
"regions": ["testland"],
"planet": planet.to_string_lossy(),
"work_dir": work.to_string_lossy(),
"through_step": "c",
})
}
/// **The nested-runtime crack is CLOSED** (issue #35): a genuinely SEEDLESS
/// server drives the REAL torrent branch — blank config planet, EMPTY disk
/// cache, a `$HOME` with NO seed — and MUST NOT panic with "cannot start a
/// runtime from within a runtime".
///
/// osm-katana's `ensure_planet_osm()` builds its OWN tokio runtime and
/// `block_on`s it. Called INLINE on this multi-thread tokio worker (as the
/// daemon did before the fix) that nests a runtime inside a runtime and
/// PANICS — so the librqbit branch had never actually executed through the
/// daemon. The fix moves the blocking torrent acquisition off the async
/// worker via `spawn_blocking` (a non-runtime thread), so osm-katana's
/// runtime is top-level.
///
/// We point the torrent at a tiny LOCAL fixture with INVALID bytes
/// (`OSM_KATANA_PLANET_TORRENT`) so the fetch fails FAST and OFFLINE — the
/// point is proving the real path reaches the download attempt WITHOUT the
/// nested-runtime panic, not downloading the 80 GiB planet. A failed fetch is
/// a `Degraded` FACT (reconcile returns `Ok`), not a hard failure.
#[cfg(feature = "osm-daemon-download")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn seedless_torrent_branch_runs_through_daemon_without_nested_runtime_panic() {
use std::sync::Mutex as StdMutex;
// env is process-global: serialize the HOME/torrent override across tests.
static ENV_LOCK: StdMutex<()> = StdMutex::new(());
let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = tmp_dir("seedless_torrent");
// ★ THE RED PROBE FOR THE REMOVED $HOME SEED.
//
// `$HOME` holds a file with EXACTLY the well-known seed name the old resolver's
// third arm looked for, and the config DECLARES no seed. The resolver must walk
// straight past it to the torrent.
//
// Under the old arm this file was found (`seed.is_file()` was the whole test —
// size was never checked), `resolve_planet` returned it, and the torrent branch was
// never entered. That is why the torrent leg went unverified for months: on the
// developer's box the fallback always won. This test is the thing that would have
// caught it, and the `$HOME` override here is no longer scaffolding — it is the probe.
let fake_home = dir.join("home");
std::fs::create_dir_all(&fake_home).unwrap();
std::fs::write(
fake_home.join(DEFAULT_SEED),
b"a file with the right NAME and nothing else",
)
.unwrap();
let disk = dir.join("disk"); // created by reconcile; left empty
// A tiny LOCAL .torrent with invalid bencode → librqbit errors FAST, offline.
let bogus_torrent = dir.join("bogus.torrent");
std::fs::write(&bogus_torrent, b"not a valid torrent file").unwrap();
// Redirect osm-katana's cache to a nonexistent temp path so it never sees a
// real >1 GiB cached planet and always attempts the (fast-failing) fetch.
let fake_cache = dir.join("cache").join("planet-latest.osm.bz2");
let prev_home = std::env::var_os("HOME");
let prev_torrent = std::env::var_os("OSM_KATANA_PLANET_TORRENT");
let prev_cache = std::env::var_os("OSM_KATANA_PLANET");
// SAFETY: guarded by ENV_LOCK; restored below before the guard drops.
unsafe {
std::env::set_var("HOME", &fake_home);
std::env::set_var("OSM_KATANA_PLANET_TORRENT", &bogus_torrent);
std::env::set_var("OSM_KATANA_PLANET", &fake_cache);
}
let (phase_sink, phases) = PhaseSink::recorder();
// Blank planet path (""), so config branch misses → cache → seed → torrent.
let cfg = serde_json::json!({
"regions": ["testland"], "planet": "", "through_step": "c",
});
let ctx = CollectorCtx {
name: NAME.into(),
disk_dir: disk.clone(),
config: cfg.clone(),
bucket: Arc::new(MemObjectStore::new()),
warehouse: Arc::new(MemWarehouse::new()),
fetch: Fetcher::noop(),
phase: phase_sink,
log: LogSink::noop(),
bios: BiosLog::new().bus(),
shutdown: Shutdown::new(),
secret: std::sync::Arc::new(crate::daemon::EnvSecretStore::new()),
marks: std::sync::Arc::new(crate::freshness::MemMarkStore::new()),
nudge: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
let daemon = OsmCutDaemon::with_config(true, cfg);
// The whole point: this call USED TO PANIC (nested runtime) the moment the
// torrent branch ran on the async worker. It must now return — Ok, with a
// Degraded fact — proving the crack is closed. A generous timeout guards
// against any librqbit hang (a hang is still not the nested-runtime panic,
// but we want the test bounded).
let res = tokio::time::timeout(
std::time::Duration::from_secs(90),
daemon.reconcile_once(&ctx),
)
.await;
// Restore env before asserting (so a panic can't leak the override).
unsafe {
match prev_home {
Some(v) => std::env::set_var("HOME", v),
None => std::env::remove_var("HOME"),
}
match prev_torrent {
Some(v) => std::env::set_var("OSM_KATANA_PLANET_TORRENT", v),
None => std::env::remove_var("OSM_KATANA_PLANET_TORRENT"),
}
match prev_cache {
Some(v) => std::env::set_var("OSM_KATANA_PLANET", v),
None => std::env::remove_var("OSM_KATANA_PLANET"),
}
}
let reconcile = res.expect("reconcile did not hang (bounded torrent attempt)");
reconcile.expect("a failed torrent is a Degraded FACT, reconcile returns Ok");
// It genuinely ENTERED the torrent branch: torrent_planet sets Downloading
// before calling into osm-katana. Seeing Downloading proves the resolver fell all
// the way through config → cache → declared seeds → TORRENT, and specifically that
// it did NOT take the file sitting in $HOME under the well-known name.
let seq = phases.lock().unwrap().clone();
assert!(
seq.iter().any(|p| matches!(p, Phase::Downloading { .. })),
"reached the TORRENT download attempt (Downloading phase): {seq:?}",
);
// …and the $HOME file was neither read nor linked anywhere under the collector's own
// directory. A resolver that quietly linked it in would still have shown Downloading.
assert!(
!disk.join(DEFAULT_SEED).exists(),
"a file in $HOME was linked into {} — the $HOME seed arm is supposed to be GONE, \
and a declared path is the only way a seed enters",
disk.display()
);
// The fast-failing offline fetch is Degraded (alive, impaired), not Online.
assert!(
matches!(daemon.phase(), Phase::Degraded { .. }),
"seedless offline torrent → Degraded fact, got {:?}",
daemon.phase(),
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **The never-ending loop is stoppable** (design §5). Pre-trip shutdown, run the
/// full daemon loop, and assert it reconciles once to Online then returns (does
/// not spin forever) — the config-toggle-off / server-shutdown path.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn daemon_loop_is_stoppable() {
let dir = tmp_dir("stop");
let planet = seed_fixture(&dir);
let shutdown = Shutdown::new();
shutdown.stop(); // config toggled off before/at start
let ctx = CollectorCtx {
name: NAME.into(),
disk_dir: dir.clone(),
config: daemon_config_for(&planet),
bucket: Arc::new(MemObjectStore::new()),
warehouse: Arc::new(MemWarehouse::new()),
fetch: Fetcher::noop(),
phase: PhaseSink::noop(),
log: LogSink::noop(),
bios: BiosLog::new().bus(),
shutdown,
secret: std::sync::Arc::new(crate::daemon::EnvSecretStore::new()),
marks: std::sync::Arc::new(crate::freshness::MemMarkStore::new()),
nudge: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
let daemon = Arc::new(OsmCutDaemon::with_config(
true,
daemon_config_for(&planet),
));
// Must return promptly (one reconcile, then the shutdown check breaks the loop).
tokio::time::timeout(std::time::Duration::from_secs(60), daemon.clone().run(ctx))
.await
.expect("stoppable daemon returns");
assert_eq!(
daemon.phase(),
Phase::Online,
"reconciled to Online before stopping"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **The region → bbox clip RESOLUTION** (the region-clip fix). The built-in
/// table resolves the common regions (Sweden ≈ lon 10.5..24.2, lat 55.0..69.1),
/// an unknown region is `None`, and a config `bbox`/`clip` override — a per-region
/// map, a bare 4-array, or a `Bounds::parse` string/preset — wins over the table.
#[test]
fn region_clip_resolution_table_and_override() {
let none_cfg = serde_json::json!({});
// 1) The built-in table: Sweden's bbox (the from-scratch #35 region).
let se = resolve_region_clip(&none_cfg, "sweden").expect("sweden is in the table");
assert_eq!(se, osm_katana::Bounds::new(10.5, 55.0, 24.2, 69.1));
// Case-insensitive.
assert_eq!(resolve_region_clip(&none_cfg, "SWEDEN"), Some(se));
// 2) An unknown region resolves to None (⇒ caller cuts the whole source).
assert_eq!(resolve_region_clip(&none_cfg, "atlantis"), None);
// 3) Config override — a bare 4-array applies to any region.
let arr_cfg = serde_json::json!({ "bbox": [1.0, 2.0, 3.0, 4.0] });
assert_eq!(
resolve_region_clip(&arr_cfg, "atlantis"),
Some(osm_katana::Bounds::new(1.0, 2.0, 3.0, 4.0)),
"a bare bbox array clips a region the table doesn't know",
);
// 4) Config override — a per-region map, looked up by region name.
let map_cfg = serde_json::json!({ "bbox": { "custom": [0.0, 0.0, 2.0, 2.0] } });
assert_eq!(
resolve_region_clip(&map_cfg, "custom"),
Some(osm_katana::Bounds::new(0.0, 0.0, 2.0, 2.0)),
);
assert_eq!(
resolve_region_clip(&map_cfg, "other"),
None,
"map miss ⇒ no clip"
);
// 5) Config override — a `Bounds::parse` string / preset (accepts `clip` too).
let preset_cfg = serde_json::json!({ "clip": "nordics" });
assert_eq!(
resolve_region_clip(&preset_cfg, "whatever"),
Some(osm_katana::Bounds::nordics())
);
// 6) An override for a KNOWN region wins over the built-in table.
let ovr_cfg = serde_json::json!({ "bbox": { "sweden": [1.0, 1.0, 2.0, 2.0] } });
assert_eq!(
resolve_region_clip(&ovr_cfg, "sweden"),
Some(osm_katana::Bounds::new(1.0, 1.0, 2.0, 2.0)),
"the config override wins over the table",
);
// 7) A malformed override (max < min, wrong arity) is rejected → table fallback.
let bad_cfg = serde_json::json!({ "bbox": { "sweden": [9.0, 9.0, 1.0, 1.0] } });
assert_eq!(
resolve_region_clip(&bad_cfg, "sweden"),
Some(se),
"bad override ⇒ table"
);
}
/// **`cut_region_to_store` CLIPS to the region bbox, dropping out-of-bbox
/// features** (the from-scratch #35 fix — a REGION cut, not a whole-planet cut).
///
/// A tiny synthetic multi-region PBF holds two clusters of nodes/ways FAR apart
/// (cluster A ≈ lon/lat 0..0.5, cluster B ≈ lon/lat 50..50.5), built from `.osm`
/// through osm-katana's own public XML→PBF converter (no committed binary). We
/// prove the clip drops cluster B two ways: (a) directly at the decode — the SAME
/// pass-1 node test `convert` runs — keeps strictly fewer nodes than an unclipped
/// decode and more than zero; (b) end-to-end — a `cut_region_to_store` clipped to
/// cluster A's bbox writes strictly FEWER (and >0) tiles than an unclipped cut of
/// the whole source. That is the difference between a Sweden-scale cut and tiling
/// the whole planet under the region's name.
#[test]
fn cut_region_clips_to_bbox_dropping_out_of_bbox_features() {
let dir = tmp_dir("region_clip");
// Two clusters, far apart so they occupy disjoint tiles at every zoom.
let osm = r#"<?xml version="1.0" encoding="UTF-8"?>
<osm version="0.6" generator="korp-collectors-test">
<node id="1" lat="0.0" lon="0.0"/>
<node id="2" lat="0.0" lon="0.5"/>
<node id="3" lat="0.5" lon="0.5"/>
<node id="4" lat="0.5" lon="0.0"/>
<node id="11" lat="50.0" lon="50.0"/>
<node id="12" lat="50.0" lon="50.5"/>
<node id="13" lat="50.5" lon="50.5"/>
<node id="14" lat="50.5" lon="50.0"/>
<way id="10">
<nd ref="1"/><nd ref="2"/><nd ref="3"/><nd ref="4"/><nd ref="1"/>
<tag k="highway" v="residential"/><tag k="name" v="Cluster A Road"/>
</way>
<way id="20">
<nd ref="11"/><nd ref="12"/><nd ref="13"/><nd ref="14"/><nd ref="11"/>
<tag k="highway" v="residential"/><tag k="name" v="Cluster B Road"/>
</way>
</osm>
"#;
let osm_path = dir.join("multiregion.osm");
std::fs::write(&osm_path, osm).unwrap();
let pbf_path = dir.join("multiregion.osm.pbf");
osm_katana::xml_to_pbf::xml_to_pbf_raw_path(&osm_path, &pbf_path, 1)
.expect("synthetic .osm → PBF via osm-katana");
// The clip for our test region "alpha" comes from a config bbox override that
// covers ONLY cluster A (lon/lat -1..2) — the exact resolution path production
// uses, exercised here for a region the built-in table does not know.
let cfg = serde_json::json!({ "bbox": { "alpha": [-1.0, -1.0, 2.0, 2.0] } });
let clip = resolve_region_clip(&cfg, "alpha").expect("override resolves the bbox");
assert_eq!(clip, osm_katana::Bounds::new(-1.0, -1.0, 2.0, 2.0));
// (a) DIRECT node-level proof: the same pass-1 decode `convert` runs keeps
// only in-bbox nodes (cluster A = 4), dropping cluster B.
let bytes = std::fs::read(&pbf_path).unwrap();
let all_nodes = osm_katana::decode_pbf_node_count(&bytes, None);
let kept_nodes =
osm_katana::decode_pbf_node_count(&bytes, Some(&osm_katana::Clip::Bbox(clip)));
assert_eq!(all_nodes, 8, "unclipped decode sees both clusters");
assert_eq!(
kept_nodes, 4,
"the bbox clip keeps ONLY cluster A's 4 nodes"
);
assert!(
kept_nodes < all_nodes && kept_nodes > 0,
"clip drops out-of-bbox, keeps in-bbox"
);
// (b) END-TO-END proof through the daemon cut: a bbox-clipped cut writes
// strictly fewer (and >0) rows than an unclipped cut of the whole source.
// Measured on the APPLIED OUTPUT — the row count in the nodes.parquet each
// cut actually produced — not on a state round-trip.
let clipped_dir = dir.join("alpha");
let whole_dir = dir.join("whole");
cut_region_extract(
&RegionCut::from_config(
&cfg,
&pbf_path,
"alpha",
&clipped_dir,
Some(osm_katana::Clip::Bbox(clip)),
),
&LogSink::noop(),
)
.expect("clipped cut");
cut_region_extract(
&RegionCut::from_config(&cfg, &pbf_path, "whole", &whole_dir, None),
&LogSink::noop(),
)
.expect("unclipped cut of the whole source");
let (clipped, _) = table_rows_and_bytes(&clipped_dir.join("nodes.parquet"));
let (whole, _) = table_rows_and_bytes(&whole_dir.join("nodes.parquet"));
assert!(
clipped > 0,
"the clipped region cut still produces rows (RAGNARÖK)"
);
assert!(
clipped < whole,
"the bbox-clipped cut ({clipped} node(s)) is a REGION extract — strictly fewer than \
converting the whole source ({whole} node(s)); a from-scratch Sweden cut is \
region-scale, not the whole planet",
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **DEFERRED OSM FIX #2 — the PRECISE polygon clip is OPT-IN and FALLS BACK
/// to the memory-safe bbox.** Over a small synthetic PBF carrying a real admin
/// relation "Alpha" (boundary=administrative + admin_level + name) that bounds
/// cluster A, [`resolve_region_cut_clip`]:
/// 1. defaults to the axis-aligned [`osm_katana::Clip::Bbox`] (no `clip_mode`),
/// 2. with `clip_mode=polygon` derives Alpha's PRECISE boundary as an
/// [`osm_katana::Clip::Poly`] — via the BOUNDED streaming subset-scan, not
/// the whole-planet in-RAM path — that CONTAINS a point inside the boundary
/// and REJECTS a far point a bbox would over-include, at boundary scale, and
/// 3. for a region with NO admin boundary in this planet, FALLS BACK to the
/// memory-safe bbox (the precise clip never regresses the default).
#[test]
fn polygon_clip_is_opt_in_and_falls_back_to_bbox() {
use crate::daemon::LogSink;
let dir = tmp_dir("polyclip");
// A real admin relation "Alpha" (two ways closing a square around cluster A,
// lon/lat 0..0.5) plus a far non-member node — built through osm-katana's own
// public XML→PBF converter (no committed binary fixture).
let osm = r#"<?xml version="1.0" encoding="UTF-8"?>
<osm version="0.6" generator="korp-collectors-test">
<node id="1" lat="0.0" lon="0.0"/>
<node id="2" lat="0.0" lon="0.5"/>
<node id="3" lat="0.5" lon="0.5"/>
<node id="4" lat="0.5" lon="0.0"/>
<node id="11" lat="50.0" lon="50.0"/>
<way id="10"><nd ref="1"/><nd ref="2"/><nd ref="3"/></way>
<way id="20"><nd ref="3"/><nd ref="4"/><nd ref="1"/></way>
<relation id="100">
<member type="way" ref="10" role="outer"/>
<member type="way" ref="20" role="outer"/>
<tag k="type" v="boundary"/>
<tag k="boundary" v="administrative"/>
<tag k="admin_level" v="4"/>
<tag k="name" v="Alpha"/>
</relation>
</osm>
"#;
let osm_path = dir.join("region.osm");
std::fs::write(&osm_path, osm).unwrap();
let pbf = dir.join("region.osm.pbf");
osm_katana::xml_to_pbf::xml_to_pbf_raw_path(&osm_path, &pbf, 1)
.expect("synthetic .osm → PBF via osm-katana");
// 1) DEFAULT clip_mode ⇒ memory-safe bbox for a KNOWN region.
let cfg_default = serde_json::json!({});
match resolve_region_cut_clip(&cfg_default, "sweden", &pbf, &LogSink::noop()) {
Some(osm_katana::Clip::Bbox(_)) => {}
other => panic!("default clip_mode must be the bbox clip, got {other:?}"),
}
// 2) OPT-IN polygon: derive Alpha's PRECISE boundary from the planet.
let cfg_poly = serde_json::json!({ "clip_mode": "polygon" });
match resolve_region_cut_clip(&cfg_poly, "alpha", &pbf, &LogSink::noop()) {
Some(osm_katana::Clip::Poly(p)) => {
assert!(
p.contains(0.25, 0.25),
"a point inside Alpha's boundary is kept"
);
assert!(
!p.contains(50.0, 50.0),
"the far point is dropped (a bbox over-includes it)"
);
assert!(
p.vertex_count() < 100,
"boundary-scale, not planet: {}",
p.vertex_count()
);
}
other => panic!("clip_mode=polygon must derive a Poly clip for Alpha, got {other:?}"),
}
// 3) FALLBACK: polygon mode for a region with NO admin boundary in this
// planet falls back to the memory-safe bbox (here Sweden's table box).
match resolve_region_cut_clip(&cfg_poly, "sweden", &pbf, &LogSink::noop()) {
Some(osm_katana::Clip::Bbox(b)) => {
assert_eq!(
b,
osm_katana::Bounds::new(10.5, 55.0, 24.2, 69.1),
"no admin polygon named 'sweden' in this planet ⇒ fall back to the bbox",
);
}
other => panic!("a missing polygon must fall back to the bbox clip, got {other:?}"),
}
let _ = std::fs::remove_dir_all(&dir);
}
// ─────────────────────────────────────────────────────────────────────────
// The four gaps a DEFAULT collector cut still had on 2026-08-02 (G1–G4).
// Each guard is written so that reverting its fix turns it RED.
// ─────────────────────────────────────────────────────────────────────────
/// Two clusters: one inside the region's real shape, one a NEIGHBOUR that sits
/// inside the region's bounding box but outside its polygon — the synthetic form
/// of Norway/Denmark/Finland inside Sweden's box. Built through osm-katana's own
/// public XML→PBF converter (no committed binary fixture).
fn poly_vs_bbox_fixture(dir: &Path) -> (PathBuf, String) {
let osm = r#"<?xml version="1.0" encoding="UTF-8"?>
<osm version="0.6" generator="korp-collectors-test">
<node id="1" lat="0.10" lon="0.10"/>
<node id="2" lat="0.10" lon="0.40"/>
<node id="3" lat="0.40" lon="0.40"/>
<node id="4" lat="0.40" lon="0.10"/>
<node id="11" lat="0.60" lon="0.60"/>
<node id="12" lat="0.60" lon="0.90"/>
<node id="13" lat="0.90" lon="0.90"/>
<node id="14" lat="0.90" lon="0.60"/>
<way id="10">
<nd ref="1"/><nd ref="2"/><nd ref="3"/><nd ref="4"/><nd ref="1"/>
The file goes on; the console shows all of it.