src/osm_cut_verify.rs
//! **`osm-cut` cut POST-CONDITIONS and SAFETY PROPERTIES** (feature `osm-daemon`).
//!
//! Everything here exists because of one failure mode: **a run that produces a wrong
//! pyramid and reports success.** On 2026-08-02 that happened repeatedly, and every
//! time the truth came from a probe an agent wrote *outside* the collector, *after*
//! the fact. A probe outside the collector cannot fail the run. These are inside it,
//! and they DO fail it — through the daemon's own
//! [`crate::daemon::Phase::Failed`]/[`crate::daemon::Phase::Degraded`], carried on the
//! existing [`crate::osm_cut::CutStats`] plumbing rather than a parallel channel.
//!
//! ## What is checked, and what each one has actually caught
//!
//! | check | what it asserts | the miss it exists for (MEASURED 2026-08-02) |
//! |---|---|---|
//! | `foreign_nodes` | every node in the extract is inside the clip, **per row group** | 16 583 foreign nodes of 96 850 300 — 0.017 %, invisible in aggregate, **98.1 % of them (16 260) in the LAST row group** |
//! | `wholly_foreign_ways` | no way lies entirely outside the clip | 1 170 of 8 612 923 in the leaky cut, 0 after |
//! | `wkb_shape` | every geometry's declared type and byte length agree | an audit reported "60.48 % wholly foreign" because its decoder read closed ways (WKB POLYGON, ~59 % of ways) as LINESTRING — 4 bytes out of phase. The real answer was 0 |
//! | `geometry_quality` | every geometry column holds SHAPES — no undecodable blob, no unclosed ring, no NEW self-crossing ring, books that close | 11 510 bowties in `relations::geometry_z9` + 99 unclosed and 599 self-crossing rings in `ways::geometry`, every one found by a human running `osm-katana verify --geometry-quality` AFTER a chain run that reported success (2026-09-02) |
//! | `present` | **named geometry actually EXISTS where it was declared to** | the check nobody had. A polygon that silently dropped Gotland, Öland and 55 % of Blekinge passed as "0 foreign" — because "0 foreign" is fidelity to the POLYGON, never to geography |
//!
//! ## Two rows were DELETED from that table, and the one above replaces them
//!
//! The refused-placements row and the tiles-per-zoom row were in the table and in
//! nothing else. Tiles-per-zoom went with the tile pyramid; refused-placements never
//! had a producer in this module — the table promised a gate that could not fail. The
//! shape of a check that cannot go red is a check that hides the ones that can.
//!
//! `the_module_names_no_check_that_does_not_exist` keeps both names out, by reading
//! this file. That is why this paragraph spells them with hyphens and not as the
//! identifiers they were: a deletion note written in the dead names is the deletion
//! undone, and the guard says so.
//!
//! ## The declaration describes what was ASKED FOR; the checks assert it ARRIVED
//!
//! Nothing here knows that the subject of a cut is "a region's whole geometry". A
//! [`Presence`] expectation names a place, a radius and how much must be there, and
//! optionally a `subject` substring the feature's tags must carry. Cutting roads (or
//! any other feature class) out later is one more declared expectation, not a new
//! check — which is why none of this is shaped around administrative regions.
//!
//! ## Cost is declared, not chosen for the operator
//!
//! The parquet probes take a [`ProbeDepth`]: `off`, `sample:<n>` (first *n* **and last
//! *n*** row groups — the head-only sample would have missed the very leak this was
//! built for), or `full`. Every outcome states which depth produced it, so a "0
//! foreign" from a sample can never be mistaken for a "0 foreign" from a full scan.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
// ─────────────────────────────────────────────────────────────────────────────
// WKB — one shape-checked walker, used by every probe.
// ─────────────────────────────────────────────────────────────────────────────
/// **What a WKB blob says it is, cross-checked against how long it actually is.**
///
/// The 2026-08-02 lesson, paid for once: a ways audit reported "60.48 % wholly
/// foreign" because its decoder read closed ways (WKB `POLYGON`, ~59 % of osm-katana's
/// ways) with the `LINESTRING` layout — a `POLYGON` carries an extra `u32` ring count
/// before the first point count, so every coordinate came out **4 bytes out of phase**
/// and the answer was confident nonsense. The real number was 0.
///
/// A decoder that cannot be wrong about this is one that computes the byte length its
/// own header implies and refuses to proceed unless it equals the blob it was handed.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct WkbShape {
/// The WKB geometry type code: 1 Point, 2 LineString, 3 Polygon.
pub kind: u32,
/// Total vertices across every ring.
pub points: usize,
/// The byte length the header implies — asserted equal to the actual blob length.
pub bytes: usize,
}
/// **Parse a WKB header and PROVE the blob is exactly as long as it claims.**
///
/// Handles the three types osm-katana writes (`Point`, `LineString`, `Polygon`;
/// tiles narrow that to `Point`/`LineString`). Anything else — a foreign type, a
/// big-endian blob, a truncated buffer, trailing bytes — is an `Err` naming the
/// disagreement, never a silent guess.
pub fn wkb_shape(w: &[u8]) -> std::result::Result<WkbShape, String> {
if w.len() < 5 {
return Err(format!(
"WKB is {} byte(s) — shorter than the 5-byte header",
w.len()
));
}
if w[0] != 1 {
return Err(format!(
"WKB byte-order byte is {} — osm-katana writes little-endian (1) only; a big-endian \
blob here means the geometry did not come from this writer",
w[0]
));
}
let u32_at = |o: usize| -> Option<u32> {
w.get(o..o + 4)
.and_then(|s| s.try_into().ok())
.map(u32::from_le_bytes)
};
let kind = u32_at(1).ok_or_else(|| "WKB type code truncated".to_string())?;
let (points, bytes) = match kind {
// 1 endian + 4 type + 2 f64
1 => (1usize, 21usize),
// 1 + 4 + 4 npoints + n*16
2 => {
let n =
u32_at(5).ok_or_else(|| "LINESTRING point count truncated".to_string())? as usize;
(n, 9 + n.saturating_mul(16))
}
// 1 + 4 + 4 nrings + per ring (4 npoints + n*16)
3 => {
let rings =
u32_at(5).ok_or_else(|| "POLYGON ring count truncated".to_string())? as usize;
let mut off = 9usize;
let mut pts = 0usize;
for r in 0..rings {
let n = u32_at(off)
.ok_or_else(|| format!("POLYGON ring {r} of {rings}: point count truncated"))?
as usize;
off = off.saturating_add(4).saturating_add(n.saturating_mul(16));
pts = pts.saturating_add(n);
}
(pts, off)
}
other => {
return Err(format!(
"WKB type {other} is not one this cut writes (1 Point / 2 LineString / 3 Polygon) \
— refusing to decode it rather than reading it out of phase"
));
}
};
if bytes != w.len() {
return Err(format!(
"WKB type {kind} declares {points} point(s) ⇒ {bytes} byte(s), but the blob is {} — \
the decoder and the writer DISAGREE, so every coordinate read from it would be out \
of phase (this is exactly the '60.48 % wholly foreign' nonsense of 2026-08-02)",
w.len()
));
}
Ok(WkbShape {
kind,
points,
bytes,
})
}
/// The vertices of a shape-checked WKB blob, in file order.
///
/// Decoding goes through [`osm_katana::shared::geometry::decode_coords`] — the reader
/// that lives beside the WRITER that produced these bytes, so the two can never drift
/// (LAW 5). [`wkb_shape`] is what makes calling it safe: it proves the blob's declared
/// layout accounts for exactly its length, so `decode_coords` cannot be reading a
/// `POLYGON` with `LINESTRING` offsets. Shape first, coordinates second — always.
pub fn wkb_points(w: &[u8]) -> std::result::Result<Vec<(f64, f64)>, String> {
let shape = wkb_shape(w)?;
let coords = osm_katana::shared::geometry::decode_coords(w)
.ok_or_else(|| format!("shape-checked WKB type {} failed to decode", shape.kind))?;
Ok(coords)
}
// ─────────────────────────────────────────────────────────────────────────────
// ProbeDepth — how deep a parquet probe goes, DECLARED not chosen.
// ─────────────────────────────────────────────────────────────────────────────
/// **How much of a parquet file a probe reads.** A full geometry scan of a Sweden-size
/// `nodes.parquet` is not free (see [`CheckOutcome::cost_ms`] — every outcome carries
/// its own measured cost), so the depth is part of the declaration.
///
/// `Sample(n)` reads the first *n* **and the last *n*** row groups, never just the
/// head: the leak this whole module exists for sat **98.1 % in the LAST row group**
/// (16 260 of 16 583), so a head-only sample is the one shape guaranteed to miss it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ProbeDepth {
/// Do not run this probe at all. Says so in the outcome — never a silent skip.
Off,
/// The first `n` and the last `n` row groups.
Sample(usize),
/// Every row group. The only depth that can report "0" without a caveat.
Full,
}
impl ProbeDepth {
/// Parse the declaration form: `false`/`"off"` ⇒ [`ProbeDepth::Off`],
/// `true`/`"full"` ⇒ [`ProbeDepth::Full`], `3`/`"sample:3"` ⇒
/// [`ProbeDepth::Sample`]. Anything unrecognised keeps `default` (a typo must not
/// silently disable a post-condition).
pub fn from_json(v: Option<&serde_json::Value>, default: ProbeDepth) -> ProbeDepth {
let Some(v) = v else { return default };
match v {
serde_json::Value::Bool(false) => ProbeDepth::Off,
serde_json::Value::Bool(true) => ProbeDepth::Full,
serde_json::Value::Number(n) => n
.as_u64()
.map(|n| ProbeDepth::Sample((n as usize).max(1)))
.unwrap_or(default),
serde_json::Value::String(s) => {
let s = s.trim().to_ascii_lowercase();
match s.as_str() {
"off" | "false" | "none" | "skip" => ProbeDepth::Off,
"full" | "true" | "all" => ProbeDepth::Full,
_ => match s.strip_prefix("sample") {
Some(rest) => {
let n = rest
.trim_matches(|c: char| !c.is_ascii_digit())
.parse::<usize>()
.unwrap_or(1);
ProbeDepth::Sample(n.max(1))
}
None => default,
},
}
}
_ => default,
}
}
/// Which of `n` row groups this depth visits.
pub fn groups(&self, n: usize) -> Vec<usize> {
match self {
ProbeDepth::Off => Vec::new(),
ProbeDepth::Full => (0..n).collect(),
ProbeDepth::Sample(k) => {
let k = (*k).max(1);
let mut v: Vec<usize> = (0..n.min(k)).collect();
v.extend(n.saturating_sub(k)..n);
v.sort_unstable();
v.dedup();
v
}
}
}
/// How this depth describes itself in an outcome — so a sampled zero is never
/// read as a proven zero.
pub fn caveat(&self, visited: usize, total: usize) -> String {
match self {
ProbeDepth::Off => "NOT RUN (depth=off)".to_string(),
ProbeDepth::Full => format!("all {total} row group(s)"),
ProbeDepth::Sample(k) => format!(
"SAMPLED {visited} of {total} row group(s) (first+last {k}) — a clean result here \
is evidence, NOT proof"
),
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The declared expectations.
// ─────────────────────────────────────────────────────────────────────────────
/// **One declared presence expectation** — "geometry must EXIST here".
///
/// The check "0 foreign" can never make: a polygon that silently drops Gotland, Öland
/// and 55 % of Blekinge is *perfectly faithful to itself* and reports 0 foreign nodes.
/// Fidelity to the polygon is not fidelity to geography, and only a declared
/// expectation about the world can tell the two apart.
///
/// `subject` is the seam for cutting things other than a whole region: when set, only
/// features whose `tags` JSON contains that substring count (`"highway"` for roads,
/// `"railway"`, …). Unset ⇒ any geometry counts.
#[derive(Clone, Debug, PartialEq)]
pub struct Presence {
/// What this place is called, for the failure message.
pub name: String,
pub lon: f64,
pub lat: f64,
/// How far around the point counts as "here". Default 25 km.
pub radius_km: f64,
/// How many features must be found. Default 1.
pub min_features: u64,
/// Which zoom to look at; `None` ⇒ the deepest zoom the run produced.
pub zoom: Option<u8>,
/// Optional tag substring a feature must carry to count (the road-cut seam).
pub subject: Option<String>,
}
impl Presence {
fn from_json(v: &serde_json::Value) -> Option<Self> {
let o = v.as_object()?;
let num = |k: &str| o.get(k).and_then(|x| x.as_f64());
Some(Self {
name: o
.get("name")
.and_then(|x| x.as_str())
.unwrap_or("unnamed place")
.to_string(),
lon: num("lon")?,
lat: num("lat")?,
radius_km: num("radius_km").unwrap_or(25.0).max(0.1),
min_features: o.get("min_features").and_then(|x| x.as_u64()).unwrap_or(1),
zoom: o
.get("zoom")
.and_then(|x| x.as_u64())
.and_then(|z| u8::try_from(z).ok()),
subject: o
.get("subject")
.and_then(|x| x.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string),
})
}
/// The lon/lat box `radius_km` around the point (a local flat-earth conversion —
/// a query box, not a geodesic, and generous by design).
#[allow(clippy::float_arithmetic)]
fn query_box(&self) -> osm_katana::Bounds {
let dlat = self.radius_km / 111.32;
let coslat = self.lat.to_radians().cos().abs().max(0.01);
let dlon = self.radius_km / (111.32 * coslat);
osm_katana::Bounds::new(
self.lon - dlon,
self.lat - dlat,
self.lon + dlon,
self.lat + dlat,
)
}
}
/// **Everything a cut declares must be true of its own output.** Parsed from the
/// collector config's `expect` key (bare, or a per-region map — the same shape every
/// other per-region declaration field uses).
#[derive(Clone, Debug, PartialEq)]
pub struct CutExpectations {
/// Depth of the per-row-group foreign-node scan of `nodes.parquet`.
pub foreign_nodes: ProbeDepth,
/// How many foreign nodes are tolerated. Default 0.
pub max_foreign_nodes: u64,
/// Depth of the wholly-foreign-way scan of `ways.parquet`.
pub foreign_ways: ProbeDepth,
/// How many wholly-foreign ways are tolerated. Default 0.
pub max_foreign_ways: u64,
/// Named places geometry must actually exist at.
pub present: Vec<Presence>,
/// Free bytes required before a planet CONVERT is started.
pub free_bytes_convert: u64,
/// **The recorded geometry-defect budget this cut is judged against.** Default
/// [`osm_katana::verify::GeometryBaseline::RECORDED`] — the live Sweden census of
/// 2026-09-02, which tolerates the 99 unclosed + 599 self-crossing rings
/// `ways::geometry` carries OUT OF CONVERT and fails on one more. Declare
/// `"geometry_baseline": "zero"` once a re-cut has cleared them (the encoder fix
/// is zoomies `10e63b8`).
pub geometry_baseline: &'static osm_katana::verify::GeometryBaseline,
}
/// The free-space floor, from the sizes MEASURED on oden 2026-08-02: a Sweden extract
/// is 5.8–6.1 GB and its `node_coords.arrow` side output ~1.5 GB. `/home` hit 100 %
/// three times that day and killed a build (`mold: failed to write to an output file.
/// Disk full?`) and a full re-cut (`No space left on device`), so the floor carries a
/// deliberate margin on top.
const DEFAULT_FREE_GB_CONVERT: f64 = 12.0;
const GB: f64 = 1_073_741_824.0;
impl Default for CutExpectations {
fn default() -> Self {
Self {
foreign_nodes: ProbeDepth::Full,
max_foreign_nodes: 0,
foreign_ways: ProbeDepth::Full,
max_foreign_ways: 0,
present: Vec::new(),
free_bytes_convert: (DEFAULT_FREE_GB_CONVERT * GB) as u64,
geometry_baseline: &osm_katana::verify::GeometryBaseline::RECORDED,
}
}
}
impl CutExpectations {
/// Read the `expect` declaration for `region` — either a bare object applied to
/// every region, or a `{"<region>": {…}}` map (the shape `poly`/`extract_dir`
/// already use, so an operator never has to learn a second spelling).
pub fn from_config(config: &serde_json::Value, region: &str) -> Self {
let mut e = Self::default();
let Some(v) = config.get("expect") else {
return e;
};
// A per-region map is one whose values are objects keyed by region name.
let node = match v.as_object() {
Some(o) if o.contains_key(region) => &o[region],
Some(o) => {
let lower = region.trim().to_ascii_lowercase();
match o.get(lower.as_str()) {
Some(x) => x,
None => v,
}
}
None => return e,
};
let Some(o) = node.as_object() else { return e };
e.foreign_nodes = ProbeDepth::from_json(o.get("foreign_nodes"), e.foreign_nodes);
e.foreign_ways = ProbeDepth::from_json(o.get("wholly_foreign_ways"), e.foreign_ways);
if let Some(n) = o.get("max_foreign_nodes").and_then(|x| x.as_u64()) {
e.max_foreign_nodes = n;
}
if let Some(n) = o.get("max_wholly_foreign_ways").and_then(|x| x.as_u64()) {
e.max_foreign_ways = n;
}
if let Some(arr) = o.get("present").and_then(|x| x.as_array()) {
e.present = arr.iter().filter_map(Presence::from_json).collect();
}
if let Some(g) = o.get("free_gb_convert").and_then(|x| x.as_f64()) {
e.free_bytes_convert = (g.max(0.0) * GB) as u64;
}
// Only "zero" moves it; an unrecognised spelling keeps the default rather
// than silently loosening the gate.
if o.get("geometry_baseline")
.and_then(|x| x.as_str())
.map(str::trim)
== Some("zero")
{
e.geometry_baseline = &osm_katana::verify::GeometryBaseline::ZERO;
}
e
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Outcomes.
// ─────────────────────────────────────────────────────────────────────────────
/// **One post-condition's verdict, with its numbers and what it cost.**
///
/// A check that only logs is not a post-condition, so `ok == false` fails the run —
/// but the numbers travel with the verdict either way, because "0 foreign nodes across
/// all 1 183 row groups in 41 s" is the sentence that makes a green trustworthy.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CheckOutcome {
/// The check's stable name (`foreign_nodes`, `present:Gotland`, …).
pub check: String,
/// Did it pass? `false` fails the whole cut.
pub ok: bool,
/// The measured numbers, in words. Always populated — a green with no numbers is
/// indistinguishable from a check that never ran.
pub detail: String,
/// Wall-clock this check cost, measured.
pub cost_ms: u128,
}
impl CheckOutcome {
fn pass(check: impl Into<String>, detail: impl Into<String>, cost_ms: u128) -> Self {
Self {
check: check.into(),
ok: true,
detail: detail.into(),
cost_ms,
}
}
fn fail(check: impl Into<String>, detail: impl Into<String>, cost_ms: u128) -> Self {
Self {
check: check.into(),
ok: false,
detail: detail.into(),
cost_ms,
}
}
/// `✔ name — detail (12 ms)`.
pub fn line(&self) -> String {
format!(
"{} {} — {} ({} ms)",
if self.ok { "✔" } else { "✘" },
self.check,
self.detail,
self.cost_ms
)
}
}
/// Every post-condition a stage ran.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CutVerdict {
pub checks: Vec<CheckOutcome>,
}
impl CutVerdict {
pub fn push(&mut self, c: CheckOutcome) {
self.checks.push(c);
}
pub fn extend(&mut self, other: CutVerdict) {
self.checks.extend(other.checks);
}
/// Total measured cost of every check in this verdict.
pub fn cost_ms(&self) -> u128 {
self.checks.iter().map(|c| c.cost_ms).sum()
}
/// `true` when at least one check RAN and every check that ran passed.
///
/// The `!is_empty()` is load-bearing, not defensive: `all()` over an empty
/// iterator is `true`, so a verdict holding zero checks used to report the same
/// word as a verdict that ran the whole suite and passed it — LAW 2's identity
/// value in its purest form (stinky `korpdep-9`). A suite that asserted nothing is
/// not a pass; `an_empty_verdict_is_not_ok` holds that red.
pub fn ok(&self) -> bool {
!self.checks.is_empty() && self.checks.iter().all(|c| c.ok)
}
/// **Checks that MEASURED but could not assert.** A pyramid with no presence
/// expectation and no calibrated hole floor may be perfectly good — nothing here
/// proves it is not — but the collector is not GREEN on that basis. The daemon
/// turns this into [`crate::daemon::Phase::Degraded`]: published, alive, and
/// honest that nothing is watching the geography.
pub fn unasserted_reason(&self) -> Option<String> {
let un: Vec<&CheckOutcome> = self
.checks
.iter()
.filter(|c| c.ok && c.detail.starts_with("UNASSERTED"))
.collect();
if un.is_empty() {
return None;
}
// The NAMES alone put the operator back in the source to learn what happened;
// the summarising check's own sentence is the one that says why, so it travels
// with the reason (the `geography_asserted` detail names a missing clip out
// loud). `failure_reason` below already carries details for the same reason.
let named = un
.iter()
.map(|c| c.check.clone())
.collect::<Vec<_>>()
.join(", ");
let why = un
.iter()
.find(|c| c.check == "geography_asserted")
.or_else(|| un.first())
.map(|c| c.detail.clone())
.unwrap_or_default();
Some(format!(
"{} post-condition(s) MEASURED but could not assert — {named}: {why}",
un.len(),
))
}
/// **The reason the run must fail**, naming every failed check and its numbers —
/// `None` when all passed. This string is what reaches
/// [`crate::daemon::Phase::Failed`], so it has to be readable by whoever is woken
/// up by it.
pub fn failure_reason(&self) -> Option<String> {
let bad: Vec<&CheckOutcome> = self.checks.iter().filter(|c| !c.ok).collect();
if bad.is_empty() {
return None;
}
Some(format!(
"{} of {} POST-CONDITION(S) FAILED — {}",
bad.len(),
self.checks.len(),
bad.iter()
.map(|c| format!("{}: {}", c.check, c.detail))
.collect::<Vec<_>>()
.join(" | ")
))
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The extract probes — foreign nodes, wholly-foreign ways, WKB shape.
// ─────────────────────────────────────────────────────────────────────────────
/// What one row group of a geometry scan found.
#[derive(Clone, Copy, Debug, Default)]
struct GroupTally {
group: usize,
rows: u64,
hits: u64,
}
/// The result of a geometry scan across the visited row groups.
struct ScanResult {
per_group: Vec<GroupTally>,
rows: u64,
hits: u64,
visited: usize,
total_groups: usize,
/// The first WKB blob whose header and length disagreed, if any.
shape_error: Option<String>,
shape_errors: u64,
}
impl ScanResult {
/// The row group holding the largest share of the hits — the sentence that turns
/// "16 583 foreign nodes" into "16 260 of them in the last row group".
fn worst_group(&self) -> Option<(usize, u64, f64)> {
let w = self.per_group.iter().max_by_key(|g| g.hits)?;
if w.hits == 0 {
return None;
}
#[allow(clippy::float_arithmetic)]
let share = (w.hits as f64) * 100.0 / (self.hits.max(1) as f64);
Some((w.group, w.hits, share))
}
}
/// **Scan the `geometry` column of a GeoParquet, row group by row group**, counting
/// the rows for which `hit` says yes — and cross-checking every blob's WKB header
/// against its byte length on the way past.
///
/// gatling fork-join, one unit per row group, LPT-scheduled heaviest-first by
/// compressed size — the SAME shape `osm_katana::verify` uses for the identical job,
/// so neither can be parallel in a way the other is not (ROOT LAW #0: gatling, never
/// rayon).
fn scan_geometry(
path: &Path,
depth: ProbeDepth,
hit: &(dyn Fn(&[(f64, f64)]) -> bool + Sync),
) -> Result<ScanResult> {
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let file = std::fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let builder = ParquetRecordBatchReaderBuilder::try_new(file)
.with_context(|| format!("read the parquet footer of {path:?}"))?;
let meta = builder.metadata().clone();
let total_groups = meta.num_row_groups();
let groups = depth.groups(total_groups);
if groups.is_empty() {
return Ok(ScanResult {
per_group: Vec::new(),
rows: 0,
hits: 0,
visited: 0,
total_groups,
shape_error: None,
shape_errors: 0,
});
}
let path = path.to_path_buf();
let meta2 = meta.clone();
let results: Vec<Result<(GroupTally, u64, Option<String>)>> =
rotaryengine::gatling_forkjoin::gatling_for_each_balanced(
groups.len(),
0,
1,
|i| meta2.row_group(groups[i]).compressed_size().max(0) as u64,
|i| scan_one_group(&path, &meta, groups[i], hit),
);
let mut per_group = Vec::with_capacity(results.len());
let (mut rows, mut hits, mut shape_errors) = (0u64, 0u64, 0u64);
let mut shape_error: Option<String> = None;
for r in results {
let (t, errs, first) = r?;
rows += t.rows;
hits += t.hits;
shape_errors += errs;
if shape_error.is_none() {
shape_error = first;
}
per_group.push(t);
}
per_group.sort_by_key(|g| g.group);
Ok(ScanResult {
per_group,
rows,
hits,
visited: groups.len(),
total_groups,
shape_error,
shape_errors,
})
}
fn scan_one_group(
path: &Path,
meta: &Arc<parquet::file::metadata::ParquetMetaData>,
group: usize,
hit: &(dyn Fn(&[(f64, f64)]) -> bool + Sync),
) -> Result<(GroupTally, u64, Option<String>)> {
use arrow_array::{Array, BinaryArray};
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::{
ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
};
// Reuse the already-parsed footer and project ONLY `geometry` — never pull the
// tags/node_refs columns through just to look at coordinates.
let arm = ArrowReaderMetadata::try_new(meta.clone(), ArrowReaderOptions::new())?;
let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(
std::fs::File::open(path).with_context(|| format!("open {path:?}"))?,
arm,
);
let mask = ProjectionMask::columns(builder.parquet_schema(), ["geometry"]);
let reader = builder
.with_row_groups(vec![group])
.with_projection(mask)
.build()?;
let mut tally = GroupTally {
group,
rows: 0,
hits: 0,
};
let mut shape_errors = 0u64;
let mut first_err: Option<String> = None;
for batch in reader {
let b = batch?;
let Some(col) = b.column(0).as_any().downcast_ref::<BinaryArray>() else {
anyhow::bail!(
"{path:?} row group {group}: the `geometry` column is {:?}, not Binary — this is \
not an osm-katana GeoParquet and no coordinate read from it would mean anything",
b.column(0).data_type()
);
};
for i in 0..col.len() {
tally.rows += 1;
if !col.is_valid(i) {
continue; // a null geometry is not a foreign one
}
match wkb_points(col.value(i)) {
Ok(coords) => {
if hit(&coords) {
tally.hits += 1;
}
}
Err(e) => {
shape_errors += 1;
if first_err.is_none() {
first_err = Some(format!("row group {group}, row {i}: {e}"));
}
}
}
}
}
Ok((tally, shape_errors, first_err))
}
/// **The tables of [`osm_katana::verify::EXTRACT_TABLES`] this cut can actually be
/// censused on** — the ones that are present AND hold at least one row group.
///
/// # Why a row-group count is read here and not left to the census
///
/// MEASURED on t14s 2026-09-06, on the `through_step: "c"` fixture cuts in
/// [`crate::osm_cut`]: a legitimate cut writes `relations.parquet` with **0 rows and 0
/// row groups**, and `osm_katana::verify::geometry_quality` fans its census over
/// `n_rg.max(1)` units — so a file with no row groups is asked for row group 0, and
/// parquet 59's `ReaderRowGroups::num_rows` indexes an empty vector: *"index out of
/// bounds: the len is 0 but the index is 0"*. It PANICS, in a worker, on healthy
/// output. Sixteen `osm_cut` tests went down with it the moment this gate was wired in.
///
/// A file with no row groups has no rows, so it can carry no defect: skipping it costs
/// the census nothing and is stated in the log rather than done quietly. This is a
/// GUARD, not a second census — the fix belongs upstream in
/// `znippy-zoomies/osm-katana/src/verify.rs` (`n_rg.max(1)` should be "no row groups ⇒
/// an empty census"), and this can go the day it lands.
fn censusable_tables(
dir: &Path,
log: &crate::daemon::LogSink,
) -> Result<Vec<osm_katana::verify::TableSpec>> {
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let mut out = Vec::with_capacity(osm_katana::verify::EXTRACT_TABLES.len());
for spec in osm_katana::verify::EXTRACT_TABLES {
let path = dir.join(spec.name);
if !path.exists() {
// Absence is `verify_dir`'s finding, not this one — two writers of the
// same fault is how two verdicts come to disagree.
continue;
}
let file =
std::fs::File::open(&path).with_context(|| format!("open {path:?} for the census"))?;
let builder = ParquetRecordBatchReaderBuilder::try_new(file)
.with_context(|| format!("read the parquet footer of {path:?}"))?;
if builder.metadata().num_row_groups() == 0 {
log.line(format!(
"post-condition geometry_quality {} holds 0 row groups — no rows, so no \
geometry to censure; SKIPPED",
spec.name
));
continue;
}
out.push(spec);
}
Ok(out)
}
/// **The SHAPES, not just their places.**
///
/// Every other post-condition here asks WHERE a geometry is. This asks whether it IS
/// one: osm-katana's own census — the one that found the 11 510 bowties — judged
/// against a recorded budget, so a known-historic defect cannot fail a cut and a NEW
/// one always does.
///
/// It takes no clip, because a broken ring is broken wherever it is. That is why it
/// runs on the no-clip arm of [`check_extract`] too: with no clip declared, every
/// other geographic question short-circuits, and this one is then the only measurement
/// the pass makes.
fn geometry_outcome(
dir: &Path,
exp: &CutExpectations,
log: &crate::daemon::LogSink,
) -> CheckOutcome {
let t = std::time::Instant::now();
let tables = match censusable_tables(dir, log) {
Ok(ts) => ts,
// Same rule as the Err arm below: a table this cut cannot open is UNVERIFIED.
Err(e) => {
return CheckOutcome::fail(
"geometry_quality",
format!("the geometry census could not read this cut: {e:#}"),
t.elapsed().as_millis(),
);
}
};
match osm_katana::verify::check_cut_geometry(dir, &tables, exp.geometry_baseline) {
Ok(g) => {
// An UNASSERTED count is carried, never dropped: `relations.resolved`'s
// base-column self-intersections were never censused on a real cut, so the
// baseline neither passes nor fails them and the operator gets to read the
// number anyway.
for n in &g.notes {
log.line(format!("post-condition geometry_quality {n}"));
}
let cost = t.elapsed().as_millis();
match g.reason() {
Some(r) => CheckOutcome::fail("geometry_quality", r, cost),
None => CheckOutcome::pass(
"geometry_quality",
format!(
"{} geometry column(s) censused against baseline `{}`, no defect \
above the record ({} UNASSERTED)",
g.census.len(),
g.baseline,
g.notes.len(),
),
cost,
),
}
}
// A census that cannot READ the tables has not proven them clean — the same
// rule `scan_outcome` applies to a probe that fails to open a file.
Err(e) => CheckOutcome::fail(
"geometry_quality",
format!("the geometry census could not read this cut: {e:#}"),
t.elapsed().as_millis(),
),
}
}
/// **Every post-condition that can be asked of the EXTRACT**, before a minute is spent
/// tiling it: foreign nodes (per row group), wholly-foreign ways, the WKB shape
/// cross-check that makes both of those numbers mean anything, and the geometry-quality
/// census that asks whether the blobs are SHAPES at all.
///
/// With no clip there is no "foreign" to be — the whole source IS the region — and the
/// probes say so rather than reporting a hollow zero.
pub fn check_extract(
dir: &Path,
clip: Option<&osm_katana::Clip>,
exp: &CutExpectations,
log: &crate::daemon::LogSink,
) -> CutVerdict {
let mut v = CutVerdict::default();
let clip_missing = clip.is_none();
let Some(clip) = clip else {
v.push(CheckOutcome::pass(
"foreign_nodes",
"UNASSERTED — no clip declared, so the whole source was cut as this region and \
nothing here can be foreign to it BY DEFINITION (NOT a scan result). The old \
wording called this a pass; it is the absence of a question, not an answer",
0,
));
v.push(CheckOutcome::pass(
"geography_asserted",
"UNASSERTED — no clip is declared for this region, so the WHOLE source was cut as \
the region: `foreign_nodes` above is a definition, not a measurement, and no \
post-condition here knows where this map is supposed to end",
0,
));
for pres in &exp.present {
let t = std::time::Instant::now();
v.push(check_presence(dir, pres, t));
}
// No clip is no excuse: the rings are still either shapes or not.
v.push(geometry_outcome(dir, exp, log));
for c in &v.checks {
log.line(format!("post-condition {}", c.line()));
}
return v;
};
// ── foreign nodes, per row group ────────────────────────────────────────
let nodes = dir.join("nodes.parquet");
let t = std::time::Instant::now();
match exp.foreign_nodes {
ProbeDepth::Off => v.push(CheckOutcome::pass(
"foreign_nodes",
"NOT RUN — depth=off in the declaration",
0,
)),
depth => {
let out = scan_geometry(&nodes, depth, &|c: &[(f64, f64)]| {
c.iter().any(|(x, y)| !clip.contains(*x, *y))
});
v.push(scan_outcome(
"foreign_nodes",
out,
depth,
exp.max_foreign_nodes,
"node(s) with a coordinate OUTSIDE the clip",
t.elapsed().as_millis(),
));
}
}
// ── wholly-foreign ways ─────────────────────────────────────────────────
let ways = dir.join("ways.parquet");
let t = std::time::Instant::now();
match exp.foreign_ways {
ProbeDepth::Off => v.push(CheckOutcome::pass(
"wholly_foreign_ways",
"NOT RUN — depth=off in the declaration",
0,
)),
depth => {
let out = scan_geometry(&ways, depth, &|c: &[(f64, f64)]| {
!c.is_empty() && c.iter().all(|(x, y)| !clip.contains(*x, *y))
});
v.push(scan_outcome(
"wholly_foreign_ways",
out,
depth,
exp.max_foreign_ways,
"way(s) with EVERY vertex outside the clip",
t.elapsed().as_millis(),
));
}
}
// ── The marker that catches a pass which asserted NOTHING about geography. ──
//
// The FIRST arm used to read `clip.is_some() && …`, which is backwards: a MISSING
// clip is when this suite knows least, not most. With `clip == None` the scans above
// short-circuit, and the marker that exists to catch a pass which asserted nothing
// was itself switched off — greens over zero measurements, and `unasserted_reason()`
// came back `None` so the daemon published ONLINE (stinky `korpdep-9`). Now a
// missing clip is its own unasserted reason, whatever else is declared.
//
// It moved here from `check_pyramid` when the pyramid was deleted. The question it
// asks was never about tiles: it is "did anything in this run know where the map is
// supposed to end", and that is a property of the CUT.
let _ = clip_missing; // the no-clip arm returned above
if exp.present.is_empty() {
v.push(CheckOutcome::pass(
"geography_asserted",
"UNASSERTED — no `present` expectation is declared, so NOTHING here asserts the \
map still has its geography. '0 foreign' is fidelity to the POLYGON, never to \
geography: a polygon that silently drops an island passes every other check on \
this list",
0,
));
}
// **The geographic post-condition.** "0 foreign" says nothing was let IN; this says
// what was declared to be here is actually here. It reads the extract, so it runs
// on every cut rather than only on a cut that produced a pyramid.
for pres in &exp.present {
let t = std::time::Instant::now();
v.push(check_presence(dir, pres, t));
}
// ── The SHAPES, not just their places. ──────────────────────────────────
v.push(geometry_outcome(dir, exp, log));
for c in &v.checks {
log.line(format!("post-condition {}", c.line()));
}
v
}
/// Turn a scan into an outcome — including the WKB shape cross-check, which is fatal
/// on its own: if the decoder and the writer disagree, the hit count is nonsense and
/// reporting it as a clean zero is the exact 2026-08-02 mistake.
fn scan_outcome(
name: &str,
out: Result<ScanResult>,
depth: ProbeDepth,
max: u64,
what: &str,
cost_ms: u128,
) -> CheckOutcome {
let out = match out {
Ok(o) => o,
// A probe that cannot READ the file has not proven the file clean.
Err(e) => {
return CheckOutcome::fail(
name,
format!(
"the probe could not run ({e:#}) — an unreadable extract is UNVERIFIED, not clean"
),
cost_ms,
);
}
};
if let Some(err) = &out.shape_error {
return CheckOutcome::fail(
name,
format!(
"WKB SHAPE CROSS-CHECK FAILED on {} of {} geometr(ies) — first: {err}. Every \
coordinate this probe read is suspect, so its count of {} is NOT a result",
out.shape_errors, out.rows, out.hits
),
cost_ms,
);
}
let scope = depth.caveat(out.visited, out.total_groups);
let worst = match out.worst_group() {
Some((g, h, share)) => format!("; worst row group {g} holds {h} of them ({share:.1} %)"),
None => String::new(),
};
let detail = format!(
"{} {what} of {} row(s) scanned across {scope}{worst}; WKB shape verified on every blob",
out.hits, out.rows
);
if out.hits > max {
CheckOutcome::fail(
name,
format!("{detail} — the declaration allows at most {max}"),
cost_ms,
)
} else {
CheckOutcome::pass(name, detail, cost_ms)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The geographic probe — presence. (It was "the pyramid probes" until the pyramid
// was deleted; the two gates named beside it here went with it — see the module
// header's deletion note.)
// ─────────────────────────────────────────────────────────────────────────────
/// **Does geometry actually EXIST at this declared place?**
///
/// Reads the extract's `ways.parquet` — the applied output this cut just produced, not
/// a state round-trip — and counts features whose geometry falls inside the
/// expectation's box. When the expectation names a `subject`, only features whose
/// `tags` carry that substring count, which is how a declaration says "there must be
/// roads here" without a new kind of check.
///
/// # It used to read the tile pyramid, and that is why it moved
///
/// The check walked `tiles/z{Z}/{X}/{Y}.parquet`. The pyramid is deleted (korp map v2,
/// `e66f846..1d53b1b`), so a check that reads it is a check that can never run — the
/// blindest possible guard. The QUESTION is unchanged and is worth keeping: a clip can
/// be perfectly faithful to its own polygon and still have dropped a place, which is
/// exactly what "0 foreign nodes" can never catch. So it asks the extract instead, and
/// asks it in one read rather than over a covering set.
fn check_presence(extract_dir: &Path, p: &Presence, started: std::time::Instant) -> CheckOutcome {
let name = format!("present:{}", p.name);
let bbox = p.query_box();
let ways = extract_dir.join("ways.parquet");
if !ways.is_file() {
return CheckOutcome::fail(
name,
format!(
"{} holds no ways.parquet, so nothing can be present anywhere",
extract_dir.display()
),
started.elapsed().as_millis(),
);
}
let found = match count_in_box(&ways, &bbox, p.subject.as_deref()) {
Ok(n) => n,
Err(e) => {
return CheckOutcome::fail(
name,
format!(
"{ways:?} could not be read ({e:#}) — an unreadable extract is UNVERIFIED, \
not empty"
),
started.elapsed().as_millis(),
);
}
};
let subject = match &p.subject {
Some(s) => format!(" carrying tag substring '{s}'"),
None => String::new(),
};
let detail = format!(
"{found} feature(s){subject} within {:.0} km of {} ({:.4},{:.4}) in {}",
p.radius_km,
p.name,
p.lon,
p.lat,
ways.display(),
);
if found < p.min_features {
CheckOutcome::fail(
name,
format!(
"{detail} — the declaration requires at least {}. Geometry declared to be here is \
MISSING: a clip can be perfectly faithful to its own polygon and still have \
dropped this place, which is why '0 foreign' could never catch it",
p.min_features
),
started.elapsed().as_millis(),
)
} else {
CheckOutcome::pass(name, detail, started.elapsed().as_millis())
}
}
/// Count the features in one parquet whose geometry has a vertex inside `bbox` (and, when
/// `subject` is set, whose tags carry that substring). Goes through the same
/// shape-checked [`wkb_points`] walker as every other probe — one decoder, one truth.
fn count_in_box(path: &Path, bbox: &osm_katana::Bounds, subject: Option<&str>) -> Result<u64> {
use arrow_array::{Array, BinaryArray, StringArray};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let file = std::fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let reader = ParquetRecordBatchReaderBuilder::try_new(file)?.build()?;
let mut n = 0u64;
for batch in reader {
let b = batch?;
let gi = b
.schema()
.index_of("geometry")
.map_err(|e| anyhow::anyhow!("{e}"))?;
let Some(geom) = b.column(gi).as_any().downcast_ref::<BinaryArray>() else {
anyhow::bail!(
"{path:?}: `geometry` is {:?}, not Binary",
b.column(gi).data_type()
);
};
let tags = b
.schema()
.index_of("tags")
.ok()
.and_then(|i| b.column(i).as_any().downcast_ref::<StringArray>().cloned());
for i in 0..geom.len() {
if !geom.is_valid(i) {
continue;
}
if let Some(want) = subject {
let has = tags
.as_ref()
.map(|t| t.is_valid(i) && t.value(i).contains(want))
.unwrap_or(false);
if !has {
continue;
}
}
let coords = wkb_points(geom.value(i)).map_err(|e| anyhow::anyhow!("{path:?}: {e}"))?;
if coords.iter().any(|(x, y)| bbox.contains(*x, *y)) {
n += 1;
}
}
}
Ok(n)
}
// ─────────────────────────────────────────────────────────────────────────────
// SAFETY PROPERTY — disk headroom before a 31–35 minute convert.
// ─────────────────────────────────────────────────────────────────────────────
thread_local! {
/// **A per-THREAD free-space override, for driving the headroom refusal RED.**
///
/// It used to be the process-global env var `KORP_TILE_FREE_BYTES`. That is a
/// shared mutable global in a test binary that runs its tests in parallel: the
/// headroom test's "1 GB free" leaked into whatever cut happened to be running
/// beside it, and refused it. The lock it held only excluded ITSELF. A cut runs
/// synchronously on its caller's thread, so a thread-local says exactly what was
/// meant — "this cut, this figure" — and cannot reach another test at all.
static FREE_BYTES_OVERRIDE: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
}
/// Pretend this thread's filesystem has `bytes` free (or `None` to stop pretending).
/// Returns the previous setting so a caller can restore it.
pub fn set_free_bytes_override(bytes: Option<u64>) -> Option<u64> {
FREE_BYTES_OVERRIDE.with(|c| c.replace(bytes))
}
/// Free bytes on the filesystem holding `path`, or `None` if it cannot be determined.
///
/// [`set_free_bytes_override`] (this thread) and then `KORP_CUT_FREE_BYTES` (the
/// operator's door) override the real figure — that is how the headroom REFUSAL is
/// driven RED end-to-end through the daemon without filling a 5.9 TB array.
pub fn free_bytes(path: &Path) -> Option<u64> {
if let Some(v) = FREE_BYTES_OVERRIDE.with(|c| c.get()) {
return Some(v);
}
if let Some(v) = std::env::var("KORP_CUT_FREE_BYTES")
.ok()
.and_then(|v| v.trim().parse::<u64>().ok())
{
return Some(v);
}
#[cfg(unix)]
{
use std::os::unix::ffi::OsStrExt;
// statvfs needs an EXISTING path; walk up to the nearest ancestor that is.
let mut probe = path;
while !probe.exists() {
probe = probe.parent()?;
}
let c = std::ffi::CString::new(probe.as_os_str().as_bytes()).ok()?;
// SAFETY: `c` is a NUL-terminated path that exists; `st` is fully written by
// statvfs on success and only read when it returned 0.
unsafe {
let mut st: libc::statvfs = std::mem::zeroed();
if libc::statvfs(c.as_ptr(), &mut st) != 0 {
return None;
}
Some((st.f_bavail as u64).saturating_mul(st.f_frsize as u64))
}
}
#[cfg(not(unix))]
{
let _ = path;
None
}
}
/// **Refuse to start heavy work that will not fit** — the property behind the three
/// times `/home` hit 100 % on 2026-08-02, killing a build (`mold: failed to write to
/// an output file. Disk full?`) and a full re-cut (`No space left on device`).
///
/// Pure, so the refusal is driven RED with two numbers and no filesystem. `free ==
/// None` (the figure could not be read) is NOT treated as "plenty": it passes with a
/// stated caveat, because refusing every run on a filesystem statvfs cannot describe
/// would be worse than proceeding — but it never claims the space was checked.
#[allow(clippy::float_arithmetic)]
pub fn headroom(free: Option<u64>, required: u64, stage: &str) -> CheckOutcome {
let gb = |b: u64| (b as f64) / GB;
match free {
None => CheckOutcome::pass(
"disk_headroom",
format!(
"free space is UNKNOWN on this filesystem — proceeding with {stage}, which needs \
{:.1} GB, WITHOUT having checked it",
gb(required)
),
0,
),
Some(f) if f < required => CheckOutcome::fail(
"disk_headroom",
format!(
"{:.1} GB free, {stage} needs {:.1} GB — REFUSING to start rather than filling \
the disk and leaving a half-written artefact behind",
gb(f),
gb(required)
),
0,
),
Some(f) => CheckOutcome::pass(
"disk_headroom",
format!("{:.1} GB free, {stage} needs {:.1} GB", gb(f), gb(required)),
0,
),
}
}
// ─────────────────────────────────────────────────────────────────────────────
// SAFETY PROPERTY — never destroy the only copy of anything.
// ─────────────────────────────────────────────────────────────────────────────
/// **Refuse to recursively delete anything that is not the collector's own scratch.**
///
/// The only irreplaceable thing on this box is the planet
/// (`planet-260427.osm.bz2`, 173 135 361 965 bytes, two copies and no third). A
/// `remove_dir_all` whose argument drifted — a `disk_dir` misconfigured to `$HOME`, a
/// region named `..` — would take it. So a recursive delete is only ever performed on
/// a path that is **strictly inside** the collector's own directory, and the check is
/// on the CANONICAL path so `../..` cannot walk out of it.
///
/// Returns the reason a delete was refused, or `None` when it is safe.
pub fn refuse_delete_outside(disk_dir: &Path, victim: &Path) -> Option<String> {
let root = disk_dir
.canonicalize()
.unwrap_or_else(|_| disk_dir.to_path_buf());
let target = victim
.canonicalize()
.unwrap_or_else(|_| victim.to_path_buf());
if target == root {
return Some(format!(
"refusing to recursively delete {target:?} — it IS the collector's own root, and a \
delete of the root is never a step in a cut"
));
}
if !target.starts_with(&root) {
return Some(format!(
"refusing to recursively delete {target:?} — it is OUTSIDE the collector's own \
directory {root:?}. Nothing outside that tree is this collector's to destroy, and \
the planet is the only copy there is"
));
}
None
}
/// [`refuse_delete_outside`] applied — a `remove_dir_all` that cannot escape the
/// collector's own tree. An absent path is `Ok` (nothing to remove).
pub fn remove_tree_guarded(disk_dir: &Path, victim: &Path) -> Result<()> {
if let Some(why) = refuse_delete_outside(disk_dir, victim) {
anyhow::bail!("{why}");
}
match std::fs::remove_dir_all(victim) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(anyhow::Error::from(e).context(format!("remove_dir_all {victim:?}"))),
}
}
// ─────────────────────────────────────────────────────────────────────────────
// SAFETY PROPERTY — rollback: a replacement that can be undone.
// ─────────────────────────────────────────────────────────────────────────────
/// **Swap a freshly-built directory into place, keeping the one it replaces.**
///
/// The mechanism, chosen from what the code already does rather than invented: build
/// into `<final>.new`, and when — and only when — every post-condition has passed,
/// `rename` the incumbent to `<final>.prev` and the new one into `<final>`. Two
/// renames on one filesystem; the window in which `<final>` does not exist is a single
/// syscall wide, and if the process dies anywhere before them, `<final>` is still the
/// last good artefact, untouched. `<final>.prev` is the rollback, and it is kept until
/// [`drop_previous`] is called after the run has published successfully.
///
/// This is what "a half-swapped extract is worse than a leaky one" costs to avoid:
/// one rename and one directory kept for the length of a run.
pub fn swap_into_place(
disk_dir: &Path,
staged: &Path,
final_path: &Path,
) -> Result<Option<PathBuf>> {
anyhow::ensure!(
staged.is_dir(),
"nothing staged at {staged:?} to swap into {final_path:?}"
);
let prev = previous_path(final_path);
// A leftover `.prev` from an earlier interrupted run must go first, or the rename
// below fails on a non-empty directory and the swap half-completes.
remove_tree_guarded(disk_dir, &prev)?;
let kept = if final_path.exists() {
std::fs::rename(final_path, &prev)
.with_context(|| format!("keep the incumbent {final_path:?} as {prev:?}"))?;
Some(prev)
} else {
None
};
if let Err(e) = std::fs::rename(staged, final_path) {
// The incumbent is still on disk under `.prev` — put it back rather than
// leaving the run with NOTHING in place. This is the rollback, executed.
if let Some(p) = &kept {
let _ = std::fs::rename(p, final_path);
}
return Err(anyhow::Error::from(e).context(format!(
"swap {staged:?} into {final_path:?} — the previous artefact has been ROLLED BACK \
into place, so nothing was lost"
)));
}
Ok(kept)
}
/// Where [`swap_into_place`] keeps the artefact it replaced.
pub fn previous_path(final_path: &Path) -> PathBuf {
let mut s = final_path.as_os_str().to_os_string();
s.push(".prev");
PathBuf::from(s)
}
/// Where a run stages an artefact it has not yet proven.
pub fn staging_path(final_path: &Path) -> PathBuf {
let mut s = final_path.as_os_str().to_os_string();
s.push(".new");
PathBuf::from(s)
}
/// Discard the kept rollback copy — called only once a run has published everything.
pub fn drop_previous(disk_dir: &Path, kept: Option<PathBuf>) {
if let Some(p) = kept {
let _ = remove_tree_guarded(disk_dir, &p);
}
}
/// Put the kept rollback copy BACK — the undo half of [`swap_into_place`], used when a
/// later stage of the same run fails after the swap already happened.
pub fn roll_back(disk_dir: &Path, final_path: &Path, kept: Option<PathBuf>) -> Result<()> {
let Some(prev) = kept else { return Ok(()) };
if !prev.is_dir() {
anyhow::bail!("nothing to roll back: {prev:?} is gone");
}
remove_tree_guarded(disk_dir, final_path)?;
std::fs::rename(&prev, final_path)
.with_context(|| format!("roll {prev:?} back into {final_path:?}"))?;
Ok(())
}
// ─────────────────────────────────────────────────────────────────────────────
// The PLANET's identity, and what makes an extract STALE.
// ─────────────────────────────────────────────────────────────────────────────
/// **What identifies the planet an extract was built from.** The owner, 2026-08-02:
/// *"filen blir 'inaktuell' när ny finns att hämta"* — an extract is stale when a NEW
/// planet exists, not when a timer expires. So staleness is a comparison of source
/// identities, and this is the identity.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PlanetIdentity {
/// The digest, or the cheap `sm-<bytes>-<mtime>` identity when no digest was taken.
pub digest: String,
/// How that digest was obtained, in words — printed so nobody has to guess whether
/// a run compared blake3 sums or file sizes.
pub source: String,
pub bytes: u64,
/// The planet's DOWNLOAD date, as the best fact available for a file already on
/// disk: its mtime. Stamped on every generation (`historisering`).
pub downloaded_unix: u64,
}
impl PlanetIdentity {
/// The one-line form that goes into the extract declaration — change it and the
/// extract is regenerated, which is exactly what "a new planet exists" must do.
pub fn stamp(&self) -> String {
format!(
"planet={}({} bytes,dl {})",
self.digest,
self.bytes,
ymd(self.downloaded_unix)
)
}
}
/// How hard to work for the planet's identity — declared, because blake3 over the
/// 173 135 361 965-byte planet is not free.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DigestMode {
/// Size + mtime only. Cheap, and still changes when a new planet is downloaded.
SizeMtime,
/// Use a published checksum sitting beside the planet if there is one, else fall
/// back to size+mtime. **The default** — the OSM planet distribution normally
/// publishes a sum next to the file, so the cheapest correct answer is to read it
/// rather than recompute 173 GB. (On oden 2026-08-02 there is NO such sidecar: the
/// only files are the two `planet-260427.osm.bz2` copies.)
Sidecar,
/// blake3 the whole file. Correct in every case and the most expensive; cached by
/// `(path, bytes, mtime)` so a steady-state tick never repeats it.
Blake3,
}
impl DigestMode {
pub fn from_json(v: Option<&serde_json::Value>) -> DigestMode {
match v
.and_then(|x| x.as_str())
.map(str::trim)
.map(str::to_ascii_lowercase)
.as_deref()
{
Some("blake3") | Some("full") => DigestMode::Blake3,
Some("size" | "mtime" | "size+mtime" | "off") => DigestMode::SizeMtime,
_ => DigestMode::Sidecar,
}
}
}
/// The checksum files a distribution might publish beside a planet, most specific
/// first. Read verbatim (first whitespace-delimited token) — the sum is the sum.
const SIDECAR_SUFFIXES: [&str; 5] = [".blake3", ".b3", ".sha256", ".sha256sum", ".md5"];
/// **Identify the planet.** Cached in `<cache_dir>/planet-identity.json` keyed by
/// path + size + mtime, so the expensive mode is paid once per downloaded planet and
/// never again — and so a NEW planet (new size or new mtime) always misses the cache.
pub fn planet_identity(
planet: &Path,
cache_dir: &Path,
mode: DigestMode,
) -> Result<PlanetIdentity> {
let md = std::fs::metadata(planet).with_context(|| format!("stat {planet:?}"))?;
let bytes = md.len();
let downloaded_unix = md
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs())
.unwrap_or(0);
let key = format!("{}|{bytes}|{downloaded_unix}|{mode:?}", planet.display());
let cache = cache_dir.join("planet-identity.json");
if let Some(hit) = std::fs::read(&cache)
.ok()
.and_then(|b| serde_json::from_slice::<serde_json::Value>(&b).ok())
.filter(|v| v.get("key").and_then(|k| k.as_str()) == Some(key.as_str()))
{
if let (Some(d), Some(src)) = (
hit.get("digest").and_then(|x| x.as_str()),
hit.get("source").and_then(|x| x.as_str()),
) {
return Ok(PlanetIdentity {
digest: d.to_string(),
source: format!("{src} (cached)"),
bytes,
downloaded_unix,
});
}
}
let size_mtime = || {
(
format!("sm-{bytes}-{downloaded_unix}"),
"size+mtime".to_string(),
)
};
let (digest, source) = match mode {
DigestMode::SizeMtime => size_mtime(),
DigestMode::Sidecar => match read_sidecar_sum(planet) {
Some((sum, file)) => (sum, format!("published checksum {file}")),
None => {
let (d, s) = size_mtime();
(d, format!("{s} (no published checksum beside the planet)"))
}
},
DigestMode::Blake3 => {
let t = std::time::Instant::now();
let d = blake3_file(planet)?;
let secs = t.elapsed().as_secs_f64().max(1e-9);
#[allow(clippy::float_arithmetic)]
let gbs = (bytes as f64) / secs / 1_073_741_824.0;
(
d,
format!("blake3 computed here ({secs:.1} s, {gbs:.2} GB/s)"),
)
}
};
let _ = std::fs::create_dir_all(cache_dir);
let _ = std::fs::write(
&cache,
serde_json::json!({ "key": key, "digest": digest, "source": source }).to_string(),
);
Ok(PlanetIdentity {
digest,
source,
bytes,
downloaded_unix,
})
}
/// A checksum published beside the planet, if there is one. Returns `(sum, filename)`.
fn read_sidecar_sum(planet: &Path) -> Option<(String, String)> {
for suf in SIDECAR_SUFFIXES {
let mut p = planet.as_os_str().to_os_string();
p.push(suf);
let p = PathBuf::from(p);
let Ok(text) = std::fs::read_to_string(&p) else {
continue;
};
let sum = text
.split_whitespace()
.next()
.unwrap_or("")
.trim()
.to_string();
if sum.len() >= 16 {
let name = p
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
return Some((sum, name));
}
}
None
}
/// blake3 of a file, streamed in 8 MiB chunks — the fleet's hash (znippy, skade), the
/// same crate at the same major, not a second hashing path (LAW 5).
pub fn blake3_file(path: &Path) -> Result<String> {
use std::io::Read;
let mut f = std::fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let mut h = blake3::Hasher::new();
let mut buf = vec![0u8; 8 << 20];
loop {
let n = f.read(&mut buf).with_context(|| format!("read {path:?}"))?;
if n == 0 {
break;
}
h.update(&buf[..n]);
}
Ok(h.finalize().to_hex().to_string())
}
/// `YYYYMMDD` from a unix timestamp — Howard Hinnant's civil-from-days, so a
/// generation can be stamped with its download date without a calendar crate.
#[allow(clippy::arithmetic_side_effects)]
pub fn ymd(unix: u64) -> String {
let z = (unix / 86_400) as i64 + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = if m <= 2 { y + 1 } else { y };
format!("{y:04}{m:02}{d:02}")
}
// ─────────────────────────────────────────────────────────────────────────────
// GENERATIONS — historisation, numbered, dated, pruned oldest-first.
// ─────────────────────────────────────────────────────────────────────────────
/// The directory name one archived generation of `final_path` gets:
/// `<final>.gen-<NNNN>-<YYYYMMDD>`. Numbered continuously (the owner: *"numrera hela
/// tiden med datumen för nedhämtning"*) and dated, so the sequence reads as history
/// rather than as a pile of `.bak`s.
pub fn generation_path(final_path: &Path, number: u64, unix: u64) -> PathBuf {
let mut s = final_path.as_os_str().to_os_string();
s.push(format!(".gen-{number:04}-{}", ymd(unix)));
PathBuf::from(s)
}
/// Every archived generation of `final_path` now on disk, oldest number first.
pub fn generations(final_path: &Path) -> Vec<(u64, PathBuf)> {
let (Some(parent), Some(stem)) = (final_path.parent(), final_path.file_name()) else {
return Vec::new();
};
let prefix = format!("{}.gen-", stem.to_string_lossy());
let Ok(rd) = std::fs::read_dir(parent) else {
return Vec::new();
};
let mut out: Vec<(u64, PathBuf)> = rd
.flatten()
.filter_map(|e| {
let name = e.file_name().to_string_lossy().to_string();
let rest = name.strip_prefix(&prefix)?;
let n: u64 = rest.split('-').next()?.parse().ok()?;
Some((n, e.path()))
})
.collect();
out.sort_by_key(|(n, _)| *n);
out
}
/// The number the NEXT generation gets — one past the highest on disk.
pub fn next_generation(final_path: &Path) -> u64 {
generations(final_path)
.last()
.map(|(n, _)| n + 1)
.unwrap_or(1)
}
/// **Swap a proven artefact into place, archiving the incumbent as a numbered,
/// dated generation, then pruning the OLDEST beyond `keep`.**
///
/// The owner, 2026-08-02: *"historisering med configbar antal, ta bort äldsta vid
/// bortag."* `keep` is the declared count of generations to retain (`keep == 0` ⇒ the
/// incumbent is archived for the length of the run and dropped by
/// [`drop_previous`], which is the no-history behaviour).
///
/// Returns the archived generation's path — the rollback target — or `None` on a
/// first-ever cut.
pub fn swap_into_place_gen(
disk_dir: &Path,
staged: &Path,
final_path: &Path,
unix: u64,
_keep: usize,
) -> Result<Option<PathBuf>> {
anyhow::ensure!(
staged.is_dir(),
"nothing staged at {staged:?} to swap into {final_path:?}"
);
let archived = if final_path.exists() {
let n = next_generation(final_path);
let slot = generation_path(final_path, n, unix);
remove_tree_guarded(disk_dir, &slot)?;
std::fs::rename(final_path, &slot)
.with_context(|| format!("archive the incumbent {final_path:?} as {slot:?}"))?;
Some(slot)
} else {
None
};
if let Err(e) = std::fs::rename(staged, final_path) {
// Put the incumbent back rather than leave NOTHING in place.
if let Some(g) = &archived {
let _ = std::fs::rename(g, final_path);
}
return Err(anyhow::Error::from(e).context(format!(
"swap {staged:?} into {final_path:?} — the previous generation has been ROLLED BACK \
into place, so nothing was lost"
)));
}
Ok(archived)
}
/// Prune archived generations down to `keep`, **oldest first**. Returns what it
/// removed. Never touches the live artefact and never escapes the collector's tree.
pub fn prune_generations(disk_dir: &Path, final_path: &Path, keep: usize) -> Vec<PathBuf> {
let gens = generations(final_path);
let mut dropped = Vec::new();
if gens.len() <= keep {
return dropped;
}
for (_, p) in gens.iter().take(gens.len() - keep) {
if remove_tree_guarded(disk_dir, p).is_ok() {
dropped.push(p.clone());
}
}
dropped
}
/// The object a READER watches to know its cache is void.
///
/// The owner, 2026-08-02: *"när nu kart generation finns och korp börjar ladda upp sin
/// första i nya generationen så skall den NUKA sin cache."* So this is PUT **before
/// the first tile of a new generation**, not after the last — the instant the served
/// pyramid stops being coherent with anything cached is the instant the upload starts,
/// and a reader that only learns at the end caches garbage for the whole upload window.
///
/// Nothing is uploaded "to korp-ui": the tiles are PUT into the S3 bucket (the
/// embedded rustfs) at `nordisk-maps/<region>/tiles/…`; korp-ui is a READER that
/// fetches from there and caches locally. This object is how the writer tells that
/// reader to drop what it has.
pub const GENERATION_FILE: &str = "collector-generation.json";
#[cfg(test)]
mod tests {
use super::*;
// ── WKB shape ───────────────────────────────────────────────────────────
fn wkb_point(x: f64, y: f64) -> Vec<u8> {
let mut b = vec![1u8];
b.extend_from_slice(&1u32.to_le_bytes());
b.extend_from_slice(&x.to_le_bytes());
b.extend_from_slice(&y.to_le_bytes());
b
}
fn wkb_line(pts: &[(f64, f64)]) -> Vec<u8> {
let mut b = vec![1u8];
b.extend_from_slice(&2u32.to_le_bytes());
b.extend_from_slice(&(pts.len() as u32).to_le_bytes());
for (x, y) in pts {
b.extend_from_slice(&x.to_le_bytes());
b.extend_from_slice(&y.to_le_bytes());
}
b
}
fn wkb_poly(ring: &[(f64, f64)]) -> Vec<u8> {
let mut b = vec![1u8];
b.extend_from_slice(&3u32.to_le_bytes());
b.extend_from_slice(&1u32.to_le_bytes());
b.extend_from_slice(&(ring.len() as u32).to_le_bytes());
for (x, y) in ring {
b.extend_from_slice(&x.to_le_bytes());
b.extend_from_slice(&y.to_le_bytes());
}
b
}
/// **The decoding trap of 2026-08-02, made impossible.** A closed way is written as
/// WKB `POLYGON`; a decoder that assumes `LINESTRING` reads it 4 bytes out of phase
/// and produced the confident nonsense "60.48 % wholly foreign" (the real answer was
/// 0). The shape check must (a) accept the polygon for exactly what it is, and (b)
/// REJECT the same bytes the moment their header and length disagree.
#[test]
fn wkb_shape_catches_the_out_of_phase_decode() {
let ring = [(11.0, 55.0), (11.1, 55.0), (11.1, 55.1), (11.0, 55.0)];
let poly = wkb_poly(&ring);
let s = wkb_shape(&poly).expect("a well-formed POLYGON passes");
assert_eq!((s.kind, s.points, s.bytes), (3, 4, poly.len()), "{s:?}");
// RED #1 — the same polygon bytes relabelled LINESTRING: the type says 2, so
// the length implied is 9 + n*16 against a body that carries an extra ring
// count. This is EXACTLY the 4-byte phase error, and it must not decode.
let mut lied = poly.clone();
lied[1..5].copy_from_slice(&2u32.to_le_bytes());
let err = wkb_shape(&lied).expect_err("a POLYGON relabelled LINESTRING must be REFUSED");
assert!(
err.contains("out of phase"),
"the message must name the fault: {err}"
);
// RED #2 — truncation.
let short = &poly[..poly.len() - 8];
assert!(
wkb_shape(short).is_err(),
"a truncated POLYGON must be refused"
);
// RED #3 — trailing junk (the header under-accounts for the blob).
let mut fat = poly.clone();
fat.push(0);
assert!(wkb_shape(&fat).is_err(), "trailing bytes must be refused");
// RED #4 — a type this writer never emits.
let mut alien = poly.clone();
alien[1..5].copy_from_slice(&7u32.to_le_bytes());
assert!(
wkb_shape(&alien).is_err(),
"an unknown WKB type must be refused, not guessed"
);
// RED #5 — big-endian.
let mut be = poly.clone();
be[0] = 0;
assert!(wkb_shape(&be).is_err(), "a big-endian blob must be refused");
// GREEN, and the coordinates come out in the RIGHT phase.
let pts = wkb_points(&poly).expect("shape-checked polygon decodes");
assert_eq!(pts, ring.to_vec(), "the exterior ring, in phase");
assert_eq!(
wkb_points(&wkb_point(13.0, 58.0)).unwrap(),
vec![(13.0, 58.0)]
);
assert_eq!(
wkb_points(&wkb_line(&[(1.0, 2.0), (3.0, 4.0)])).unwrap(),
vec![(1.0, 2.0), (3.0, 4.0)]
);
}
/// **The phase error, quantified.** Reading the polygon with LINESTRING offsets
/// does not merely shift bytes — it invents coordinates far outside any clip, which
/// is how "0 foreign" became "60.48 % wholly foreign". Proves the wrong reading is
/// genuinely wrong (so the guard has something real to catch) AND that the right
/// reading is right.
#[test]
fn the_out_of_phase_reading_really_does_invent_foreign_coordinates() {
let ring = [(18.0, 57.0), (18.1, 57.0), (18.1, 57.1), (18.0, 57.0)];
let poly = wkb_poly(&ring);
let sweden = osm_katana::Bounds::new(10.5, 55.0, 24.2, 69.1);
// In-phase: every vertex is in Sweden.
let good = wkb_points(&poly).unwrap();
assert!(
good.iter().all(|(x, y)| sweden.contains(*x, *y)),
"{good:?}"
);
// Out-of-phase, exactly as the bad audit read it: skip the ring count, take
// points from offset 9 instead of 13.
let mut bad = Vec::new();
let mut o = 9usize;
while o + 16 <= poly.len() {
bad.push((
f64::from_le_bytes(poly[o..o + 8].try_into().unwrap()),
f64::from_le_bytes(poly[o + 8..o + 16].try_into().unwrap()),
));
o += 16;
}
assert!(
bad.iter().all(|(x, y)| !sweden.contains(*x, *y)),
"the out-of-phase read must produce coordinates OUTSIDE Sweden — otherwise this test \
cannot tell a working decoder from a broken one: {bad:?}"
);
}
// ── ProbeDepth ──────────────────────────────────────────────────────────
/// A sample reads the FIRST n and the LAST n row groups. The leak that started all
/// of this sat 98.1 % in the LAST row group, so a head-only sample is the one
/// shape that would have passed it.
#[test]
fn sample_depth_reads_the_tail_not_only_the_head() {
assert_eq!(ProbeDepth::Full.groups(5), vec![0, 1, 2, 3, 4]);
assert_eq!(ProbeDepth::Off.groups(5), Vec::<usize>::new());
assert_eq!(ProbeDepth::Sample(2).groups(10), vec![0, 1, 8, 9]);
assert_eq!(
ProbeDepth::Sample(9).groups(3),
vec![0, 1, 2],
"a sample wider than the file"
);
assert!(
ProbeDepth::Sample(1).groups(1183).contains(&1182),
"the LAST row group is always visited — the 16 260-of-16 583 group",
);
}
#[test]
fn probe_depth_parses_the_declaration_and_a_typo_never_disables_a_check() {
let j = serde_json::json!({"a": false, "b": true, "c": "sample:3", "d": 4, "e": "wat"});
let g = |k: &str| ProbeDepth::from_json(j.get(k), ProbeDepth::Full);
assert_eq!(g("a"), ProbeDepth::Off);
assert_eq!(g("b"), ProbeDepth::Full);
assert_eq!(g("c"), ProbeDepth::Sample(3));
assert_eq!(g("d"), ProbeDepth::Sample(4));
assert_eq!(
g("e"),
ProbeDepth::Full,
"an unrecognised value keeps the default — a typo \
must never silently turn a post-condition OFF"
);
// A sampled result labels itself so its zero is never read as a proof.
assert!(ProbeDepth::Sample(2).caveat(4, 100).contains("NOT proof"));
assert!(ProbeDepth::Full.caveat(100, 100).contains("all 100"));
}
// ── headroom ────────────────────────────────────────────────────────────
/// **RED first.** A convert that does not fit must be REFUSED, by name and with
/// both numbers — this is the guard for the three times `/home` hit 100 % on
/// 2026-08-02.
#[test]
fn headroom_refuses_a_convert_that_will_not_fit() {
let need = (12.0 * GB) as u64;
let red = headroom(Some((3.0 * GB) as u64), need, "the planet convert");
assert!(
!red.ok,
"3 GB free for a 12 GB convert must FAIL: {}",
red.detail
);
assert!(red.detail.contains("3.0 GB free"), "{}", red.detail);
assert!(red.detail.contains("12.0 GB"), "{}", red.detail);
assert!(red.detail.contains("REFUSING"), "{}", red.detail);
// GREEN, and it is a different verdict for a different input (not a constant).
assert!(headroom(Some((40.0 * GB) as u64), need, "the planet convert").ok);
// Exactly at the floor is enough; one byte under is not.
assert!(headroom(Some(need), need, "x").ok);
assert!(!headroom(Some(need - 1), need, "x").ok);
// Unknown free space passes, but SAYS it did not check.
let unknown = headroom(None, need, "x");
assert!(unknown.ok && unknown.detail.contains("WITHOUT having checked"));
}
/// The real `statvfs` path reports a plausible figure for a real directory, and the
/// env override drives it RED end-to-end without filling a 5.9 TB array.
#[test]
fn free_bytes_reads_a_real_filesystem_and_can_be_overridden() {
let d = std::env::temp_dir();
let real = free_bytes(&d).expect("statvfs answers for a real directory");
assert!(real > 0, "a mounted filesystem reports non-zero free space");
// A path that does not exist yet still answers (it walks up to its parent).
assert!(free_bytes(&d.join("no-such-dir-3f8024eb")).is_some());
}
// ── never destroy the only copy ─────────────────────────────────────────
/// **RED first.** The recursive delete refuses anything outside the collector's own
/// tree — including the directory the planet lives in, which holds the only two
/// copies of a 173 135 361 965-byte file.
#[test]
fn a_recursive_delete_cannot_escape_the_collectors_own_tree() {
let root = std::env::temp_dir().join(format!("okv_del_{}", std::process::id()));
let inside = root.join("regions").join("sweden").join("tiles");
std::fs::create_dir_all(&inside).unwrap();
std::fs::write(inside.join("t.parquet"), b"x").unwrap();
// RED — a sibling of the collector dir (stand-in for $HOME with the planet in it).
let outside = std::env::temp_dir().join(format!("okv_outside_{}", std::process::id()));
std::fs::create_dir_all(&outside).unwrap();
std::fs::write(outside.join("planet-260427.osm.bz2"), b"the only copy").unwrap();
let why = refuse_delete_outside(&root, &outside).expect("an OUTSIDE path must be refused");
assert!(why.contains("OUTSIDE"), "{why}");
assert!(
remove_tree_guarded(&root, &outside).is_err(),
"and the guarded delete must fail"
);
assert!(
outside.join("planet-260427.osm.bz2").is_file(),
"the only copy is STILL THERE after a refused delete",
);
// RED — the collector root itself is never a delete target.
assert!(refuse_delete_outside(&root, &root).is_some());
// RED — `..` cannot walk out (the check is on the CANONICAL path).
let escape = inside.join("..").join("..").join("..").join("..");
assert!(
refuse_delete_outside(&root, &escape).is_some(),
"`..` must not escape"
);
// GREEN — a path strictly inside is removed.
assert!(refuse_delete_outside(&root, &inside).is_none());
remove_tree_guarded(&root, &inside).unwrap();
assert!(
!inside.exists(),
"the tile tree inside the collector dir IS removable"
);
// Absent is idempotent, not an error.
remove_tree_guarded(&root, &inside).unwrap();
let _ = std::fs::remove_dir_all(&root);
let _ = std::fs::remove_dir_all(&outside);
}
// ── rollback ────────────────────────────────────────────────────────────
/// **Replacing an artefact is recoverable.** Build into `.new`, swap, keep the
/// incumbent as `.prev`, and be able to put it back — proven by rolling a
/// replacement back and finding the ORIGINAL bytes, not the new ones.
#[test]
fn a_swap_keeps_the_incumbent_and_can_be_rolled_back() {
let root = std::env::temp_dir().join(format!("okv_swap_{}", std::process::id()));
let live = root.join("tiles");
std::fs::create_dir_all(&live).unwrap();
std::fs::write(live.join("t.parquet"), b"OLD").unwrap();
let staged = staging_path(&live);
std::fs::create_dir_all(&staged).unwrap();
std::fs::write(staged.join("t.parquet"), b"NEW").unwrap();
let kept = swap_into_place(&root, &staged, &live).expect("swap");
assert_eq!(std::fs::read(live.join("t.parquet")).unwrap(), b"NEW");
let prev = kept.clone().expect("the incumbent was KEPT, not destroyed");
assert_eq!(
std::fs::read(prev.join("t.parquet")).unwrap(),
b"OLD",
"the replaced artefact is still readable — that is the rollback",
);
// Roll back: the ORIGINAL bytes come back.
roll_back(&root, &live, kept).expect("roll back");
assert_eq!(
std::fs::read(live.join("t.parquet")).unwrap(),
b"OLD",
"after a rollback the run is serving exactly what it served before",
);
assert!(!prev.exists(), "the rollback consumed the kept copy");
// A swap with nothing staged is refused rather than emptying the live path.
let err = swap_into_place(&root, &staged, &live).expect_err("nothing staged");
assert!(format!("{err:#}").contains("nothing staged"), "{err:#}");
assert!(
live.join("t.parquet").is_file(),
"the live artefact survived the refusal"
);
let _ = std::fs::remove_dir_all(&root);
}
/// A first-ever cut has no incumbent — the swap must still work and report that
/// there is nothing to roll back to.
#[test]
fn a_first_swap_has_no_incumbent_to_keep() {
let root = std::env::temp_dir().join(format!("okv_swap1_{}", std::process::id()));
let live = root.join("tiles");
let staged = staging_path(&live);
std::fs::create_dir_all(&staged).unwrap();
std::fs::write(staged.join("t.parquet"), b"NEW").unwrap();
let kept = swap_into_place(&root, &staged, &live).expect("swap");
assert!(kept.is_none(), "nothing was replaced");
assert_eq!(std::fs::read(live.join("t.parquet")).unwrap(), b"NEW");
drop_previous(&root, kept);
let _ = std::fs::remove_dir_all(&root);
}
// ── expectations parsing ────────────────────────────────────────────────
#[test]
fn expectations_parse_bare_and_per_region_and_default_to_never_empty() {
// The default: every probe FULL, nothing tolerated, and a pyramid may never
// be empty even when nothing at all was declared.
let d = CutExpectations::from_config(&serde_json::json!({}), "sweden");
assert_eq!(d.foreign_nodes, ProbeDepth::Full);
assert_eq!(d.max_foreign_nodes, 0);
assert!(d.present.is_empty());
let cfg = serde_json::json!({
"expect": {
"sweden": {
"foreign_nodes": "sample:2",
"free_gb_convert": 20.0,
"present": [
{ "name": "Gotland", "lon": 18.49, "lat": 57.47, "radius_km": 40 },
{ "name": "E4 at Gävle", "lon": 17.14, "lat": 60.67, "subject": "highway",
"min_features": 5 }
]
}
}
});
let e = CutExpectations::from_config(&cfg, "sweden");
assert_eq!(e.foreign_nodes, ProbeDepth::Sample(2));
assert_eq!(e.free_bytes_convert, (20.0 * GB) as u64);
assert_eq!(e.present.len(), 2);
assert_eq!(e.present[1].subject.as_deref(), Some("highway"));
assert_eq!(e.present[1].min_features, 5);
// A region the map does not mention gets the defaults, not sweden's.
let other = CutExpectations::from_config(&cfg, "norway");
assert_eq!(other.foreign_nodes, ProbeDepth::Full);
assert!(other.present.is_empty(), "the defaults, not sweden's");
// The geometry budget: RECORDED unless a re-cut has earned `zero`, and a
// spelling nobody recognises must never LOOSEN it — the failure mode of a
// gate read out of config is a typo that switches it off.
assert_eq!(
d.geometry_baseline,
&osm_katana::verify::GeometryBaseline::RECORDED,
"an undeclared cut is judged against the record, not against nothing"
);
let z = CutExpectations::from_config(
&serde_json::json!({ "expect": { "geometry_baseline": " zero " } }),
"sweden",
);
assert_eq!(z.geometry_baseline, &osm_katana::verify::GeometryBaseline::ZERO);
let typo = CutExpectations::from_config(
&serde_json::json!({ "expect": { "geometry_baseline": "none" } }),
"sweden",
);
assert_eq!(
typo.geometry_baseline,
&osm_katana::verify::GeometryBaseline::RECORDED,
"an unrecognised spelling keeps the default; it does not disable the gate"
);
}
// ── the verdict ─────────────────────────────────────────────────────────
#[test]
fn a_verdict_names_which_check_failed_and_with_what_numbers() {
let mut v = CutVerdict::default();
v.push(CheckOutcome::pass("wholly_foreign_ways", "0 of 8612923", 1));
assert!(v.ok() && v.failure_reason().is_none());
v.push(CheckOutcome::fail(
"foreign_nodes",
"16583 foreign of 96850300",
41_000,
));
let r = v
.failure_reason()
.expect("a failed check MUST produce a reason");
assert!(r.contains("foreign_nodes"), "{r}");
assert!(
r.contains("16583"),
"the numbers travel with the reason: {r}"
);
assert!(r.contains("1 of 2"), "{r}");
assert_eq!(v.cost_ms(), 41_001);
}
// ── the geometry-quality gate in the chain (row 39) ──────────────────────
/// Write the ONE column every probe here reads: an osm-katana-shaped GeoParquet
/// with a nullable Binary `geometry`. Nothing else is needed — `scan_geometry`
/// projects `geometry` and so does the census.
fn write_geometry_table(path: &Path, rows: &[Option<Vec<u8>>]) {
use arrow_array::RecordBatch;
use arrow_array::builder::BinaryBuilder;
use arrow_schema::{DataType, Field, Schema};
use parquet::arrow::ArrowWriter;
let mut b = BinaryBuilder::new();
for r in rows {
match r {
Some(v) => b.append_value(v),
None => b.append_null(),
}
}
let schema = Arc::new(Schema::new(vec![Field::new(
"geometry",
DataType::Binary,
true,
)]));
let batch = RecordBatch::try_new(schema, vec![Arc::new(b.finish())]).expect("batch");
let f = std::fs::File::create(path).expect("create the fixture table");
let mut w = ArrowWriter::try_new(f, batch.schema(), None).expect("writer");
w.write(&batch).expect("write");
w.close().expect("close");
}
/// **A cut whose rings are BROKEN must not pass the chain's own verify** (row 39).
///
/// RED before the `geometry_quality` post-condition existed: every check
/// `check_extract` had asked WHERE a geometry is — inside the clip, at a declared
/// place, with a header that matches its length — and none asked whether it IS a
/// shape. The cut below is built so every one of those is green BY CONSTRUCTION
/// (nothing foreign, every WKB well-formed) while carrying an unclosed ring and a
/// bowtie. On 2026-09-02 that is exactly what the live Sweden run reported: four
/// tables, all green, over 99 open rings and 599 crossing ones.
#[test]
fn a_planted_bad_ring_makes_the_chains_own_verify_refuse_the_cut() {
let dir = std::env::temp_dir().join(format!("okv_geomgate_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("fixture dir");
let clip = osm_katana::Clip::Bbox(osm_katana::Bounds::new(10.0, 54.0, 12.0, 56.0));
write_geometry_table(
&dir.join("nodes.parquet"),
&[Some(wkb_point(11.0, 55.0)), Some(wkb_point(11.1, 55.1))],
);
// row 0 a clean closed square; row 1 the SAME square with its closing vertex
// dropped (unclosed ring); row 2 a closed, asymmetric bowtie (4 segments, two
// of them properly crossing).
let square = [
(11.0, 55.0),
(11.2, 55.0),
(11.2, 55.2),
(11.0, 55.2),
(11.0, 55.0),
];
let open = [(11.0, 55.0), (11.2, 55.0), (11.2, 55.2), (11.0, 55.2)];
let bowtie = [
(11.0, 55.0),
(11.2, 55.2),
(11.2, 55.0),
(11.0, 55.4),
(11.0, 55.0),
];
write_geometry_table(
&dir.join("ways.parquet"),
&[
Some(wkb_poly(&square)),
Some(wkb_poly(&open)),
Some(wkb_poly(&bowtie)),
],
);
// `"zero"` because a three-row fixture is not the 2026-09-02 Sweden cut: the
// RECORDED baseline tolerates 99 unclosed + 599 crossing rings in
// `ways::geometry`, so a planted pair would sit UNDER the record and prove
// nothing about the gate.
let exp = CutExpectations::from_config(
&serde_json::json!({ "expect": { "geometry_baseline": "zero" } }),
"sweden",
);
let v = check_extract(&dir, Some(&clip), &exp, &crate::daemon::LogSink::noop());
let seen = v
.checks
.iter()
.map(CheckOutcome::line)
.collect::<Vec<_>>()
.join("\n");
assert!(
!v.ok(),
"a cut carrying an unclosed ring and a bowtie passed EVERY post-condition — \
the chain asks where a geometry is and never whether it is a shape:\n{seen}"
);
let r = v
.failure_reason()
.expect("a refusal names the check that produced it");
assert!(r.contains("geometry_quality"), "{r}");
assert!(
r.contains("unclosed_ring"),
"the numbers travel with the refusal: {r}"
);
assert!(r.contains("self_intersecting"), "{r}");
let _ = std::fs::remove_dir_all(&dir);
}
/// **A table with 0 row groups is skipped, not crashed on** (row 39).
///
/// RED when the gate was first wired in: 16 `osm_cut` tests died at
/// `parquet-59.2.0/src/arrow/arrow_reader/mod.rs:1307` — *"index out of bounds: the
/// len is 0 but the index is 0"* — because a `through_step: "c"` cut writes
/// `relations.parquet` with 0 rows and 0 row groups and the census fans over
/// `n_rg.max(1)`. Healthy output, panicking gate. See [`censusable_tables`].
#[test]
fn an_empty_table_is_skipped_by_the_census_instead_of_panicking_it() {
let dir = std::env::temp_dir().join(format!("okv_geomempty_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("fixture dir");
write_geometry_table(&dir.join("ways.parquet"), &[Some(wkb_poly(&[
(11.0, 55.0),
(11.2, 55.0),
(11.2, 55.2),
(11.0, 55.0),
]))]);
// 0 rows ⇒ parquet writes 0 row groups. This is what the chain really emits.
write_geometry_table(&dir.join("relations.parquet"), &[]);
let exp = CutExpectations::from_config(
&serde_json::json!({ "expect": { "geometry_baseline": "zero" } }),
"sweden",
);
let (log, lines) = crate::daemon::LogSink::recorder();
let c = geometry_outcome(&dir, &exp, &log);
assert!(c.ok, "a clean way table and an EMPTY relation table is a PASS: {}", c.line());
assert!(
lines
.lock()
.expect("the recorder")
.iter()
.any(|l| l.contains("relations.parquet") && l.contains("0 row groups")),
"the skip is stated, never quiet: {:?}",
lines.lock().expect("the recorder")
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **No post-condition names a gate that cannot fail** (row 39).
///
/// The names are split across `concat!` on purpose: this test reads its own file,
/// and a whole literal here would match itself.
#[test]
fn the_module_names_no_check_that_does_not_exist() {
let src = include_str!("osm_cut_verify.rs");
for dead in [concat!("refused_", "placements"), concat!("tiles_per_", "zoom")] {
assert!(
!src.contains(dead),
"`{dead}` is named here and produced nowhere — a gate that cannot go \
red hides the gates that can. Delete it or give it a producer",
);
}
}
}