osm-katana/src/provenance.rs
//! **What an extract was cut FROM — the record that keys a work directory to a planet
//! (row 38, 2026-09-05).**
//!
//! # The defect this closes
//!
//! korp's `pipeline::coords_state` and `tables_state` decide whether step `c` is done by
//! asking whether the parquet files are **complete**: `PAR1` at both ends, and nothing
//! else. So the chain answers *"already converted"* for a work directory produced from a
//! **different planet file**. Point the pipeline at a newer
//! `planet-latest.osm.bz2` and every table cut from the old one is accepted as the new
//! one's output. Nothing anywhere goes wrong; the served map is simply months out of
//! date and reports itself current — the same false-Current shape
//! [`crate::spatial::PackRecipe`] closes one file over, and the same one the `Other`
//! type-spelling collision had.
//!
//! # Why this is not [`crate::coords_resume::CoordsProvenance`]
//!
//! That record already carries the source's path, length and mtime, and it is the right
//! record for what it does — it authorises reusing **`node_coords.arrow`**, and it is
//! written beside that artefact. But the coordinate store is 1.66 GB of derived scratch
//! and is swept; the TABLES outlive it, and nothing described *them*. So this is a
//! second record with a different lifetime, sharing the one [`SourceStamp`] so the two
//! cannot disagree about what identifies a source (LAW 5).
//!
//! # The digest MODE is part of the record, and that is the point
//!
//! Length and mtime miss a re-download that kept both — the case that actually happens
//! when a mirror re-publishes. A digest closes it, but a full digest of a 70 GB planet
//! is not always affordable, so the record **states which check was made** rather than
//! implying one. A reader that knows it is trusting [`DigestMode::HeadTail`] knows
//! exactly what that is worth; a reader trusting an unlabelled "digest" does not.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
/// Name of the record written into an extract directory by `convert`.
pub const EXTRACT_PROVENANCE_FILE: &str = "extract.provenance";
/// How many bytes each of head and tail [`DigestMode::HeadTail`] reads.
///
/// 8 MiB either end: enough to cover a PBF header block, the string tables of the first
/// blobs and the tail of the last, and cheap enough to run on a 70 GB planet without
/// anyone deciding to skip it. A check that is skipped in practice is worth nothing,
/// which is the argument for a partial digest over an unaffordable exact one.
pub const HEAD_TAIL_BYTES: u64 = 8 * 1024 * 1024;
/// **Which check a [`SourceStamp`]'s digest actually made.**
///
/// Recorded, never implied. The three are not interchangeable and a reader is entitled
/// to know which one it is trusting.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DigestMode {
/// No content was read. Length and mtime only — the check a re-download that keeps
/// its name and size passes.
None,
/// The first and last [`HEAD_TAIL_BYTES`], plus the length. Catches a truncation, a
/// re-cut and a different file; **does not** catch an edit confined to the middle,
/// and says so.
HeadTail,
/// Every byte. Exact, and priced accordingly.
Full,
}
impl DigestMode {
/// The spelling that goes in the record.
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
DigestMode::None => "none",
DigestMode::HeadTail => "headtail",
DigestMode::Full => "full",
}
}
/// The inverse. `None` for a spelling this build does not know — which a reader must
/// treat as "cannot judge", never as agreement.
#[must_use]
pub fn from_str(s: &str) -> Option<DigestMode> {
match s {
"none" => Some(DigestMode::None),
"headtail" => Some(DigestMode::HeadTail),
"full" => Some(DigestMode::Full),
_ => None,
}
}
}
/// **What identifies a source file** — the ONE answer, shared by this record and
/// [`crate::coords_resume::CoordsProvenance`].
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SourceStamp {
/// Absolute path as the run saw it.
pub path: String,
/// Length in bytes. Catches a truncated or re-downloaded file that kept its name.
pub len: u64,
/// mtime in nanoseconds since the epoch. Catches a same-size rewrite — usually.
pub mtime_ns: u128,
/// Which check [`Self::digest`] is the result of.
pub digest_mode: DigestMode,
/// The digest, hex, or empty under [`DigestMode::None`].
pub digest: String,
}
impl SourceStamp {
/// Stamp `path` with the named check.
pub fn of(path: &Path, mode: DigestMode) -> Result<Self> {
let md = std::fs::metadata(path)
.with_context(|| format!("stat source {}", path.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 {
path: path
.canonicalize()
.unwrap_or_else(|_| path.to_path_buf())
.to_string_lossy()
.into_owned(),
len: md.len(),
mtime_ns,
digest_mode: mode,
digest: digest_of(path, mode, md.len())?,
})
}
/// **What differs between this stamp and `other`, named** — or `None` when nothing
/// does.
///
/// Ordered cheapest-first so the reason a reader is shown is the one it would have
/// reached anyway. The PATH is deliberately last: a work directory moved between
/// mounts is the least interesting difference and the most common false alarm, so it
/// is reported only when the bytes agree.
#[must_use]
pub fn difference(&self, other: &Self) -> Option<String> {
if self.len != other.len {
return Some(format!(
"source len differs: the extract was cut from {} bytes, this source is {}",
self.len, other.len
));
}
if self.digest_mode == other.digest_mode
&& self.digest_mode != DigestMode::None
&& self.digest != other.digest
{
return Some(format!(
"source digest differs ({}): the extract was cut from {}, this source is {}",
self.digest_mode.as_str(),
self.digest,
other.digest
));
}
if self.digest_mode != other.digest_mode {
return Some(format!(
"source was stamped with the `{}` check and is being judged with `{}` — \
the two cannot be compared",
self.digest_mode.as_str(),
other.digest_mode.as_str()
));
}
if self.mtime_ns != other.mtime_ns {
return Some(format!(
"source mtime differs: cut from {}, this source is {}",
self.mtime_ns, other.mtime_ns
));
}
if self.path != other.path {
return Some(format!(
"source path differs: cut from {}, this source is {}",
self.path, other.path
));
}
None
}
}
/// **The record an extract directory carries.**
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ExtractProvenance {
/// The source this extract was cut from.
pub source: SourceStamp,
/// Signature of the active region clip, or `none` — a clip changes which rows
/// survive, so a differently clipped cut of one planet is a different extract.
pub clip_sig: String,
/// The `osm-katana` version that cut it, for a human reading the file.
pub katana_version: String,
}
/// What an extract directory says about the source it is being judged against.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SourceVerdict {
/// The record describes this very source.
Current,
/// The record describes a different one. Carries the axis that differs.
Changed(String),
/// There is no record. **Not** `Current`: silence is not agreement, and every
/// extract cut before row 38 is in this state.
Absent(String),
}
impl ExtractProvenance {
fn path(dir: &Path) -> PathBuf {
dir.join(EXTRACT_PROVENANCE_FILE)
}
/// Describe the convert that just finished.
pub fn of(source: &Path, clip_sig: String, mode: DigestMode) -> Result<Self> {
Ok(Self {
source: SourceStamp::of(source, mode)?,
clip_sig,
katana_version: String::from(env!("CARGO_PKG_VERSION")),
})
}
/// Write the record into the extract directory.
///
/// Written **last**, after every table is durable, so a record that exists implies
/// the tables it describes finished — the same ordering rule
/// [`crate::coords_resume::CoordsProvenance::write`] follows, and for the same
/// reason: absence must read as "cannot trust", which is where a torn run has to
/// land.
pub fn write(&self, dir: &Path) -> Result<()> {
let body = format!(
"source={}\nsource_len={}\nsource_mtime_ns={}\ndigest_mode={}\ndigest={}\n\
clip_sig={}\nkatana_version={}\n",
self.source.path,
self.source.len,
self.source.mtime_ns,
self.source.digest_mode.as_str(),
self.source.digest,
self.clip_sig,
self.katana_version,
);
let p = Self::path(dir);
std::fs::write(&p, body).with_context(|| format!("write {}", p.display()))
}
/// Read a record, or `None` when it is absent or unparseable. An unparseable record
/// is treated exactly like a missing one: it cannot authorise anything.
#[must_use]
pub fn read(dir: &Path) -> Option<Self> {
let text = std::fs::read_to_string(Self::path(dir)).ok()?;
let (mut path, mut len, mut mtime, mut mode, mut digest) = (None, None, None, None, None);
let (mut clip, mut ver) = (None, None);
for line in text.lines() {
let (k, v) = line.split_once('=')?;
match k {
"source" => path = Some(v.to_string()),
"source_len" => len = v.parse().ok(),
"source_mtime_ns" => mtime = v.parse().ok(),
"digest_mode" => mode = DigestMode::from_str(v),
"digest" => digest = Some(v.to_string()),
"clip_sig" => clip = Some(v.to_string()),
"katana_version" => ver = Some(v.to_string()),
_ => {}
}
}
Some(Self {
source: SourceStamp {
path: path?,
len: len?,
mtime_ns: mtime?,
digest_mode: mode?,
digest: digest?,
},
clip_sig: clip?,
katana_version: ver?,
})
}
/// **Is the extract in `dir` the output of `source`?**
///
/// The ten-line answer korp's `coords_state` / `tables_state` need, so neither of
/// them has to know how a source is identified.
pub fn verdict(dir: &Path, source: &Path) -> Result<SourceVerdict> {
let Some(rec) = Self::read(dir) else {
return Ok(SourceVerdict::Absent(format!(
"{dir:?} carries no {EXTRACT_PROVENANCE_FILE} — nothing on disk says \
which source produced these tables, so a completeness check accepts \
them as the output of any planet. Re-run `convert` to key them."
)));
};
let now = SourceStamp::of(source, rec.source.digest_mode)?;
Ok(match rec.source.difference(&now) {
Some(why) => SourceVerdict::Changed(why),
None => SourceVerdict::Current,
})
}
}
/// The digest of `path` under `mode`.
///
/// FNV-1a, the hash [`crate::digest`] already uses for row identity — one hash in the
/// crate, not two. It is not a cryptographic digest and does not claim to be: the
/// question here is "did this file change", not "did an adversary change it".
fn digest_of(path: &Path, mode: DigestMode, len: u64) -> Result<String> {
use std::io::{Read, Seek, SeekFrom};
if mode == DigestMode::None {
return Ok(String::new());
}
let mut f = std::fs::File::open(path)
.with_context(|| format!("open source {}", path.display()))?;
let mut h = crate::digest::RowHasher::new();
// The length rides in the hash so a truncation cannot be masked by the head and the
// tail happening to line up.
h.write(&len.to_le_bytes());
let mut buf = vec![0u8; 1024 * 1024];
match mode {
DigestMode::None => unreachable!("returned above"),
DigestMode::Full => {
loop {
let n = f.read(&mut buf)?;
if n == 0 {
break;
}
h.write(&buf[..n]);
}
}
DigestMode::HeadTail => {
let span = HEAD_TAIL_BYTES.min(len);
let mut left = span;
while left > 0 {
let want = (left as usize).min(buf.len());
let n = f.read(&mut buf[..want])?;
if n == 0 {
break;
}
h.write(&buf[..n]);
left -= n as u64;
}
// The tail, and only when there IS one that the head did not already cover.
if len > span {
f.seek(SeekFrom::Start(len - span))?;
let mut left = span;
while left > 0 {
let want = (left as usize).min(buf.len());
let n = f.read(&mut buf[..want])?;
if n == 0 {
break;
}
h.write(&buf[..n]);
left -= n as u64;
}
}
}
}
Ok(format!("{:016x}", h.finish()))
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ **RED-when-broken: the head/tail digest reads neither the head nor the tail.**
///
/// A digest that hashed only the length would satisfy every "did it change" arm that
/// changes the length, which is most of them. Each region is moved on its own.
#[test]
fn the_head_tail_digest_reads_both_ends_and_says_what_it_missed() {
let d = tempfile::tempdir().unwrap();
let f = d.path().join("s.bin");
let n = (HEAD_TAIL_BYTES * 3) as usize;
let base = vec![5u8; n];
std::fs::write(&f, &base).unwrap();
let stamp = SourceStamp::of(&f, DigestMode::HeadTail).unwrap();
for (what, at) in [("head", 0usize), ("tail", n - 1)] {
let mut b = base.clone();
b[at] = 9;
std::fs::write(&f, &b).unwrap();
assert_ne!(
SourceStamp::of(&f, DigestMode::HeadTail).unwrap().digest,
stamp.digest,
"a changed byte in the {what} did not move the head/tail digest"
);
}
// The DECLARED limitation, asserted rather than left as a comment: a change in
// the middle is invisible to this mode. That is exactly why the mode is written
// into the record instead of an unlabelled "digest".
let mut b = base.clone();
b[n / 2] = 9;
std::fs::write(&f, &b).unwrap();
assert_eq!(
SourceStamp::of(&f, DigestMode::HeadTail).unwrap().digest,
stamp.digest,
"the head/tail digest saw a change in the middle — either it is reading the \
whole file (and is mispriced) or the fixture is smaller than the spans"
);
assert_ne!(
SourceStamp::of(&f, DigestMode::Full).unwrap().digest,
SourceStamp::of(&f, DigestMode::Full).unwrap().digest.clone() + "x",
"sanity"
);
// …and Full does see it.
std::fs::write(&f, &base).unwrap();
let full = SourceStamp::of(&f, DigestMode::Full).unwrap();
std::fs::write(&f, &b).unwrap();
assert_ne!(
SourceStamp::of(&f, DigestMode::Full).unwrap().digest,
full.digest,
"the FULL digest missed a change in the middle"
);
}
/// ★ **RED-when-broken: `difference` reports nothing, or reports the wrong axis.**
///
/// Each axis is moved on its own, and the ORDER is asserted — a reader shown "path
/// differs" when the bytes also differ would chase a moved mount instead of a
/// changed planet.
#[test]
fn every_axis_of_a_source_stamp_is_named_when_it_moves() {
let base = SourceStamp {
path: String::from("/planet/a.osm.pbf"),
len: 100,
mtime_ns: 7,
digest_mode: DigestMode::HeadTail,
digest: String::from("abc"),
};
assert_eq!(base.difference(&base.clone()), None, "a stamp differs from itself");
let mut len = base.clone();
len.len = 101;
assert!(base.difference(&len).unwrap().contains("len"));
let mut dig = base.clone();
dig.digest = String::from("def");
assert!(base.difference(&dig).unwrap().contains("digest"));
let mut mt = base.clone();
mt.mtime_ns = 8;
assert!(base.difference(&mt).unwrap().contains("mtime"));
let mut p = base.clone();
p.path = String::from("/planet/b.osm.pbf");
assert!(base.difference(&p).unwrap().contains("path"));
// ORDER: when the bytes AND the path differ, the bytes are what is reported.
let mut both = base.clone();
both.len = 101;
both.path = String::from("/planet/b.osm.pbf");
let why = base.difference(&both).unwrap();
assert!(
why.contains("len") && !why.contains("path differs"),
"a changed planet was reported as a moved path: {why}"
);
// Comparing across MODES is refused rather than answered — a head/tail digest
// and a full one are different numbers of the same file, and reporting that as
// "the source changed" would be a lie.
let mut m = base.clone();
m.digest_mode = DigestMode::Full;
assert!(base.difference(&m).unwrap().contains("cannot be compared"));
}
/// The record survives a round trip through the file, and an unparseable one reads
/// as absent rather than as anything else.
#[test]
fn the_record_round_trips_and_a_damaged_one_is_absent() {
let d = tempfile::tempdir().unwrap();
let f = d.path().join("s.bin");
std::fs::write(&f, b"hello").unwrap();
let rec = ExtractProvenance::of(&f, String::from("none"), DigestMode::Full).unwrap();
rec.write(d.path()).unwrap();
assert_eq!(ExtractProvenance::read(d.path()).as_ref(), Some(&rec));
assert_eq!(
ExtractProvenance::verdict(d.path(), &f).unwrap(),
SourceVerdict::Current
);
// A record missing a field cannot authorise anything — the same answer as no
// record at all, which is what a partially written file has to land on.
std::fs::write(d.path().join(EXTRACT_PROVENANCE_FILE), "source=/x\nsource_len=5\n")
.unwrap();
assert_eq!(ExtractProvenance::read(d.path()), None);
assert!(matches!(
ExtractProvenance::verdict(d.path(), &f).unwrap(),
SourceVerdict::Absent(_)
));
}
}