osm-katana/src/coords_resume.rs
//! Resuming a 2-pass convert at pass 2, without re-running pass 1.
//!
//! Pass 1 of the resolved readers exists to produce exactly two artefacts:
//! `node_coords.arrow` (the sorted, mmap-backed node store pass 2 resolves way
//! and relation geometry through) and — when nodes are included —
//! `nodes.parquet`. `NodeStore::finalize` leaves the Arrow file on disk and
//! rebuilds the store from it with `coords_ipc::open_mmap` +
//! `NodeStore::from_arrow_mmap`. A convert that already has those artefacts can
//! therefore skip pass 1 and the k-way merge entirely and open the store
//! directly. On the Sweden extract that is 1 365 s of the wall not paid twice.
//!
//! **An artefact is reused only when it can prove what produced it.** Existence is
//! not proof: a `node_coords.arrow` cut from another extract, or from this extract
//! under a different region clip, is a valid Arrow file holding a different node
//! set, and pass 2 would resolve geometry against it and write parquet that reads
//! as correct everywhere downstream. So a provenance record sits beside the
//! artefact, and reuse requires an exact match against the convert that is asking.
//!
//! The tiling recipe carries the same shape — a claim written next to the data and
//! compared before the data is trusted. Any artefact that outlives the run which
//! made it earns one.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use crate::Clip;
use crate::node_store::NodeStore;
/// Name of the provenance sidecar written next to `node_coords.arrow`.
pub const PROVENANCE_FILE: &str = "node_coords.provenance";
/// Arrow IPC files begin and end with this magic. A file still being written
/// has no trailing copy, which is what makes it a completeness marker and not
/// merely a format check.
const ARROW_MAGIC: &[u8] = b"ARROW1";
/// Parquet's trailing magic, same role for `nodes.parquet`.
const PARQUET_MAGIC: &[u8] = b"PAR1";
/// What produced a `node_coords.arrow`.
///
/// Compared field-for-field before the store is reused. Every field is here
/// because changing it changes the node set the file holds.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CoordsProvenance {
/// Absolute path of the source extract, as the convert saw it.
pub input: String,
/// Source length in bytes. Catches a re-downloaded or truncated extract
/// that kept its name.
pub input_len: u64,
/// Source mtime in nanoseconds since the epoch. Catches a same-size rewrite.
pub input_mtime_ns: u128,
/// Signature of the active region clip, or `none`. A clip changes which
/// nodes survive pass 1, so a store cut under a different clip is a
/// different store.
pub clip_sig: String,
/// Whether pass 1 also wrote `nodes.parquet`. A store built without it
/// cannot satisfy a convert that wants node records.
pub include_nodes: bool,
/// Node count the producing run finished with, checked against the count the
/// Arrow footer reports on reopen.
pub node_count: usize,
}
impl CoordsProvenance {
/// Describe the convert that is about to run (or has just run).
pub fn of(
input: &Path,
clip: Option<&Clip>,
include_nodes: bool,
node_count: usize,
) -> Result<Self> {
let md =
std::fs::metadata(input).with_context(|| format!("stat input {}", input.display()))?;
let mtime_ns = md
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_nanos())
.unwrap_or(0);
Ok(Self {
input: input
.canonicalize()
.unwrap_or_else(|_| input.to_path_buf())
.to_string_lossy()
.into_owned(),
input_len: md.len(),
input_mtime_ns: mtime_ns,
clip_sig: clip
.map(clip_signature)
.unwrap_or_else(|| "none".to_string()),
include_nodes,
node_count,
})
}
fn path(dir: &Path) -> PathBuf {
dir.join(PROVENANCE_FILE)
}
/// Write the record beside the artefact it describes.
///
/// Written **after** the Arrow file is durable, so a record that exists implies
/// the artefact it names finished. Absence of a record reads as "cannot reuse",
/// which is the verdict a torn run must land on.
pub fn write(&self, dir: &Path) -> Result<()> {
let body = format!(
"input={}\ninput_len={}\ninput_mtime_ns={}\nclip_sig={}\ninclude_nodes={}\nnode_count={}\n",
self.input,
self.input_len,
self.input_mtime_ns,
self.clip_sig,
self.include_nodes,
self.node_count,
);
let path = Self::path(dir);
std::fs::write(&path, body).with_context(|| format!("write {}", path.display()))
}
/// Read a record, or `None` if it is absent or unparseable. An unparseable
/// record is treated exactly like a missing one: it cannot authorise reuse.
pub fn read(dir: &Path) -> Option<Self> {
let text = std::fs::read_to_string(Self::path(dir)).ok()?;
let mut input = None;
let mut input_len = None;
let mut input_mtime_ns = None;
let mut clip_sig = None;
let mut include_nodes = None;
let mut node_count = None;
for line in text.lines() {
let (k, v) = line.split_once('=')?;
match k {
"input" => input = Some(v.to_string()),
"input_len" => input_len = v.parse().ok(),
"input_mtime_ns" => input_mtime_ns = v.parse().ok(),
"clip_sig" => clip_sig = Some(v.to_string()),
"include_nodes" => include_nodes = v.parse().ok(),
"node_count" => node_count = v.parse().ok(),
_ => {}
}
}
Some(Self {
input: input?,
input_len: input_len?,
input_mtime_ns: input_mtime_ns?,
clip_sig: clip_sig?,
include_nodes: include_nodes?,
node_count: node_count?,
})
}
}
/// A stable signature for a clip: everything that changes which nodes survive.
///
/// FNV-1a over the discriminant, the bounds bits, and — for a polygon — every
/// ring vertex. Polygons here are region outlines (hundreds of vertices, not
/// millions), so hashing all of them costs nothing next to a convert and avoids
/// the "same bbox, different outline" collision a bounds-only signature has.
pub fn clip_signature(clip: &Clip) -> String {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
let mut eat = |bytes: &[u8]| {
for b in bytes {
h ^= *b as u64;
h = h.wrapping_mul(0x1000_0000_01b3);
}
};
let b = clip.bounds();
match clip {
Clip::Bbox(_) => eat(b"bbox"),
Clip::Poly(_) => eat(b"poly"),
}
for v in [b.min_lon, b.min_lat, b.max_lon, b.max_lat] {
eat(&v.to_le_bytes());
}
if let Clip::Poly(p) = clip {
for ring in p.rings() {
eat(&(ring.len() as u64).to_le_bytes());
for (lon, lat) in ring {
eat(&lon.to_le_bytes());
eat(&lat.to_le_bytes());
}
}
}
format!("{h:016x}")
}
/// True if `path` both starts and ends with `magic`.
///
/// Both ends, because a file mid-write already carries the leading copy. The
/// trailing one is written last, so it is the marker that separates a finished
/// artefact from one still being filled.
fn has_magic_both_ends(path: &Path, magic: &[u8]) -> bool {
use std::io::{Read, Seek, SeekFrom};
let Ok(mut f) = std::fs::File::open(path) else {
return false;
};
let Ok(len) = f.metadata().map(|m| m.len()) else {
return false;
};
if len < (magic.len() as u64) * 2 {
return false;
}
let mut head = vec![0u8; magic.len()];
if f.read_exact(&mut head).is_err() || head != magic {
return false;
}
if f.seek(SeekFrom::End(-(magic.len() as i64))).is_err() {
return false;
}
let mut tail = vec![0u8; magic.len()];
f.read_exact(&mut tail).is_ok() && tail == magic
}
/// Parquet has no leading magic requirement for our purposes beyond `PAR1`,
/// which it does carry at both ends.
fn parquet_complete(path: &Path) -> bool {
has_magic_both_ends(path, PARQUET_MAGIC)
}
/// Why a resume was refused. Returned so the caller can say it out loud rather
/// than silently falling back to a 1 365 s pass 1 that looks like normal work.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ResumeVerdict {
/// No provenance record beside the artefact.
NoProvenance,
/// The record describes a different convert. Carries a short reason.
Mismatch(String),
/// `node_coords.arrow` is missing, truncated, or has no trailing magic.
ArrowIncomplete,
/// `nodes.parquet` is required by this convert but is missing or truncated.
NodesParquetIncomplete,
/// The Arrow footer's node count disagrees with the record.
CountDisagrees { recorded: usize, found: usize },
}
impl std::fmt::Display for ResumeVerdict {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::NoProvenance => write!(f, "no {PROVENANCE_FILE} beside the artefact"),
Self::Mismatch(why) => write!(f, "provenance describes a different convert: {why}"),
Self::ArrowIncomplete => write!(f, "node_coords.arrow missing or incomplete"),
Self::NodesParquetIncomplete => write!(f, "nodes.parquet missing or incomplete"),
Self::CountDisagrees { recorded, found } => {
write!(
f,
"node count disagrees: provenance says {recorded}, Arrow footer says {found}"
)
}
}
}
}
/// Open the node store left by a previous pass 1, if and only if every artefact
/// that pass owns is present, complete, and provably from **this** convert.
///
/// Returns `Ok(store)` on a clean resume, or `Err(verdict)` naming what stopped
/// it. Never returns a store built from a partial file: an incomplete artefact
/// and an absent one are the same answer here, which is the whole point.
pub fn reusable_store(
out_dir: &Path,
input: &Path,
clip: Option<&Clip>,
include_nodes: bool,
) -> std::result::Result<(NodeStore, usize), ResumeVerdict> {
let Some(recorded) = CoordsProvenance::read(out_dir) else {
return Err(ResumeVerdict::NoProvenance);
};
let wanted = CoordsProvenance::of(input, clip, include_nodes, recorded.node_count)
.map_err(|e| ResumeVerdict::Mismatch(format!("cannot stat input: {e}")))?;
if recorded.input != wanted.input {
return Err(ResumeVerdict::Mismatch(format!(
"input was {}, now {}",
recorded.input, wanted.input
)));
}
if recorded.input_len != wanted.input_len {
return Err(ResumeVerdict::Mismatch(format!(
"input length was {}, now {}",
recorded.input_len, wanted.input_len
)));
}
if recorded.input_mtime_ns != wanted.input_mtime_ns {
return Err(ResumeVerdict::Mismatch("input mtime changed".to_string()));
}
if recorded.clip_sig != wanted.clip_sig {
return Err(ResumeVerdict::Mismatch(format!(
"clip was {}, now {}",
recorded.clip_sig, wanted.clip_sig
)));
}
// A store cut without node records cannot serve a convert that wants them;
// the reverse is fine — the parquet is simply already there and unused.
if include_nodes && !recorded.include_nodes {
return Err(ResumeVerdict::Mismatch(
"previous run did not write nodes.parquet".to_string(),
));
}
let arrow = out_dir.join("node_coords.arrow");
if !has_magic_both_ends(&arrow, ARROW_MAGIC) {
return Err(ResumeVerdict::ArrowIncomplete);
}
if include_nodes && !parquet_complete(&out_dir.join("nodes.parquet")) {
return Err(ResumeVerdict::NodesParquetIncomplete);
}
let mc = crate::coords_ipc::open_mmap(&arrow).map_err(|_| ResumeVerdict::ArrowIncomplete)?;
if mc.count != recorded.node_count {
return Err(ResumeVerdict::CountDisagrees {
recorded: recorded.node_count,
found: mc.count,
});
}
let store = NodeStore::from_arrow_mmap(mc.mmap, mc.id_off, mc.lat_off, mc.lon_off, mc.count);
Ok((store, recorded.node_count))
}
#[cfg(test)]
mod tests {
//! Every test here asserts a **refusal**, because the failure this module
//! prevents is a resume that is taken when it should not be — pass 2
//! resolving geometry against the wrong node store writes plausible, wrong
//! parquet, and nothing downstream can tell. A test that only proved the
//! happy path would pass just as well against a function that always
//! returned the store.
use super::*;
use crate::node_store::CoordFileWriter;
/// Build a complete artefact set in `dir` for `input`: a real, finalized
/// `node_coords.arrow` plus the provenance record that authorises reusing it.
fn produce(dir: &Path, input: &Path, clip: Option<&Clip>, include_nodes: bool) -> usize {
let mut w = CoordFileWriter::new(&dir.join("node_coords.bin")).expect("coord writer");
for i in 1..=1000i64 {
w.push(i, 59.0 + (i as f32) * 1e-5, 18.0 + (i as f32) * 1e-5)
.expect("push");
}
let store = w.finalize().expect("finalize");
let n = store.len();
assert!(
n > 0,
"fixture produced an empty store — the test would prove nothing"
);
if include_nodes {
std::fs::write(dir.join("nodes.parquet"), b"PAR1\x00\x00PAR1").expect("stub parquet");
}
CoordsProvenance::of(input, clip, include_nodes, n)
.expect("provenance")
.write(dir)
.expect("write provenance");
n
}
/// The refusal, or a panic naming the store that should not have been
/// returned. `NodeStore` has no `Debug`, and giving it one just to call
/// `unwrap_err` would be a production trait added for a test.
fn refusal(r: std::result::Result<(NodeStore, usize), ResumeVerdict>) -> ResumeVerdict {
match r {
Ok((_, n)) => panic!("expected a refusal, but a store of {n} nodes was reused"),
Err(v) => v,
}
}
fn an_input(dir: &Path, bytes: &[u8]) -> PathBuf {
let p = dir.join("extract.osm.bz2");
std::fs::write(&p, bytes).expect("write input");
p
}
#[test]
fn a_matching_artefact_set_is_reused_and_reports_the_producing_run_s_count() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"the same extract");
let n = produce(d.path(), &input, None, true);
let (store, count) = reusable_store(d.path(), &input, None, true)
.expect("a complete, matching artefact set must be reusable");
assert_eq!(
count, n,
"resume reported a different node count than pass 1 produced"
);
assert_eq!(
store.len(),
n,
"the reopened store holds a different number of nodes"
);
}
#[test]
fn a_missing_provenance_record_refuses_even_though_the_arrow_is_perfect() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"the same extract");
produce(d.path(), &input, None, true);
std::fs::remove_file(d.path().join(PROVENANCE_FILE)).unwrap();
// The Arrow file is untouched and completely valid. Existence is not
// provenance: this is the case where "does the file exist" would say yes.
assert_eq!(
refusal(reusable_store(d.path(), &input, None, true)),
ResumeVerdict::NoProvenance,
);
}
#[test]
fn a_different_extract_of_the_same_length_refuses_on_mtime() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"extract number one");
produce(d.path(), &input, None, true);
// Same path, same length, different bytes — the shape a re-download or a
// regenerated cut actually takes. Length alone would wave this through.
std::thread::sleep(std::time::Duration::from_millis(10));
std::fs::write(&input, b"extract number two").unwrap();
assert_eq!(std::fs::metadata(&input).unwrap().len(), 18);
match refusal(reusable_store(d.path(), &input, None, true)) {
ResumeVerdict::Mismatch(why) => assert!(
why.contains("mtime"),
"expected the mtime check to catch a same-length rewrite, got: {why}"
),
other => panic!("a rewritten input must not be reusable, got {other:?}"),
}
}
#[test]
fn a_store_cut_under_a_different_clip_refuses() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"the same extract");
let stockholm = Clip::Bbox(crate::Bounds::new(17.8, 59.2, 18.3, 59.5));
produce(d.path(), &input, Some(&stockholm), true);
// Same input, same file, different region. The node set differs; the
// Arrow file is valid and holds the wrong nodes.
let goteborg = Clip::Bbox(crate::Bounds::new(11.8, 57.6, 12.1, 57.8));
match refusal(reusable_store(d.path(), &input, Some(&goteborg), true)) {
ResumeVerdict::Mismatch(why) => assert!(
why.contains("clip"),
"expected the clip check to fire, got: {why}"
),
other => panic!("a different clip must not be reusable, got {other:?}"),
}
// …and the original clip still is, so the check discriminates rather
// than simply always refusing.
assert!(reusable_store(d.path(), &input, Some(&stockholm), true).is_ok());
}
#[test]
fn two_polygons_with_the_same_bbox_do_not_share_a_signature() {
// A bounds-only signature would call these equal and reuse a store cut
// under the wrong outline.
let square = crate::Polygon::new(vec![vec![
(0.0, 0.0),
(1.0, 0.0),
(1.0, 1.0),
(0.0, 1.0),
(0.0, 0.0),
]])
.expect("square");
let triangle =
crate::Polygon::new(vec![vec![(0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 0.0)]])
.expect("triangle");
assert_eq!(
square.bbox(),
triangle.bbox(),
"fixture must share a bbox to be the test it claims"
);
assert_ne!(
clip_signature(&Clip::Poly(square)),
clip_signature(&Clip::Poly(triangle)),
"same bbox, different outline — the signature must tell them apart",
);
}
#[test]
fn an_arrow_file_truncated_past_its_trailing_magic_refuses() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"the same extract");
produce(d.path(), &input, None, true);
// Exactly what a crash mid-write leaves: leading magic present, trailing
// magic gone. A front-only check calls this complete.
let arrow = d.path().join("node_coords.arrow");
let len = std::fs::metadata(&arrow).unwrap().len();
let f = std::fs::OpenOptions::new()
.write(true)
.open(&arrow)
.unwrap();
f.set_len(len - 7).unwrap();
assert_eq!(
refusal(reusable_store(d.path(), &input, None, true)),
ResumeVerdict::ArrowIncomplete,
);
}
#[test]
fn a_truncated_nodes_parquet_refuses_only_when_nodes_are_wanted() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"the same extract");
produce(d.path(), &input, None, true);
std::fs::write(d.path().join("nodes.parquet"), b"PAR1\x00\x00trunc").unwrap();
assert_eq!(
refusal(reusable_store(d.path(), &input, None, true)),
ResumeVerdict::NodesParquetIncomplete,
"a convert that wants node records cannot resume past a truncated nodes.parquet",
);
// Without --nodes the file is not this convert's concern.
assert!(reusable_store(d.path(), &input, None, false).is_ok());
}
#[test]
fn a_provenance_claiming_more_nodes_than_the_arrow_holds_refuses() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"the same extract");
let n = produce(d.path(), &input, None, true);
// The record and the artefact disagree — the shape of a provenance
// written for one run and an Arrow left by another.
let mut prov = CoordsProvenance::read(d.path()).unwrap();
prov.node_count = n + 1;
prov.write(d.path()).unwrap();
assert_eq!(
refusal(reusable_store(d.path(), &input, None, true)),
ResumeVerdict::CountDisagrees {
recorded: n + 1,
found: n
},
);
}
#[test]
fn a_run_without_node_records_cannot_serve_one_that_wants_them() {
let d = tempfile::tempdir().unwrap();
let input = an_input(d.path(), b"the same extract");
produce(d.path(), &input, None, false);
match refusal(reusable_store(d.path(), &input, None, true)) {
ResumeVerdict::Mismatch(why) => assert!(why.contains("nodes.parquet"), "got: {why}"),
other => panic!("expected a mismatch on missing node records, got {other:?}"),
}
}
}