osm-katana/tests/convert_cancel.rs
//! **A convert that is asked to stop, stops — and leaves nothing behind.**
//!
//! `osm_katana::convert` measures 2 665 s on a Sweden cut against a collector
//! `refresh_secs` of 3 600, so before the cancellation seam a daemon that owned
//! one was unstoppable for most of its refresh interval. These are the guards on
//! the two properties that makes it operable:
//!
//! 1. **It actually stops.** Asserted against the phase log — the applied
//! output of the run, not a flag round-trip. A convert cancelled during
//! pass 1 must never open pass 2, and the outer `convert` phase must close
//! `ok:false` naming the phase and the work-unit count where it stopped.
//! 2. **It leaves no half-written extract.** Asserted against the file system
//! at the worst moment available: after a parquet writer has already put
//! row groups on disk but before any footer exists. The output directory
//! must contain no `nodes/ways/relations.parquet` at all — the collector's
//! `extract.new` → rename staging can then never be handed a truncated file
//! carrying a complete file's name.
//!
//! Both are checked against a REAL convert of a REAL `.osm.bz2` through
//! `read_resolved_bz2` — the exact path a Sweden cut runs — not a stand-in.
//!
//! Fixture: `lbzip2/test_data/liechtenstein.osm.bz2`, 5.2 MB compressed spanning
//! many bz2 blocks (≈ 60 MB of XML), so pass 1 and pass 2 each see hundreds of
//! gatling work units. Every number in this file is a FIXTURE number; the Sweden
//! figures quoted above are from `scratch/clipleak/recut.jsonl`, 2026-08-02.
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use osm_katana::{CancelToken, ConvertOptions, is_cancelled_error};
fn fixture_bz2() -> PathBuf {
PathBuf::from(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../lbzip2/test_data/liechtenstein.osm.bz2"
))
}
const PARQUET: [&str; 3] = ["nodes.parquet", "ways.parquet", "relations.parquet"];
fn opts(dir: &Path, workers: usize, cancel: Option<CancelToken>) -> ConvertOptions {
ConvertOptions {
output_dir: dir.to_path_buf(),
log_path: Some(dir.join("phases.jsonl")),
// Keep the intermediates so their removal is observable rather than
// conflated with the `keep_node_coords=false` housekeeping path.
keep_node_coords: true,
vtd_workers: workers,
cancel,
..Default::default()
}
}
/// Workers for the cancellation guards. ONE, deliberately: it is the same code
/// path (gatling always splits, dispatches and collects — `n_workers` only sets
/// how many segments a chunk is cut into), but it stretches the fixture's passes
/// from ~0.4 s to ~1.1 s each, so the window between "the watcher sees parquet
/// bytes" and "the phase ends on its own" is not a coin flip on a loaded box.
/// The convert being cancelled is `read_resolved_bz2` either way.
const CANCEL_WORKERS: usize = 1;
/// Every parquet product still present in `dir`, with its size.
fn parquet_left(dir: &Path) -> Vec<(String, u64)> {
PARQUET
.iter()
.filter_map(|n| {
std::fs::metadata(dir.join(n))
.ok()
.map(|m| ((*n).to_string(), m.len()))
})
.collect()
}
fn phase_log(dir: &Path) -> String {
std::fs::read_to_string(dir.join("phases.jsonl")).unwrap_or_default()
}
/// `true` if the log records that phase `p` was ever entered.
fn phase_started(log: &str, p: &str) -> bool {
log.lines()
.any(|l| l.contains(&format!("\"phase\":\"{p}\"")) && l.contains("\"ev\":\"start\""))
}
/// What the watcher saw when it pulled the trigger.
struct Fired {
/// The trigger condition really held (rather than the budget running out).
real: bool,
/// When `cancel()` was called — the zero of the stop-latency measurement.
at: Instant,
/// Parquet products present in the output dir at that instant, with sizes.
on_disk: Vec<(String, u64)>,
}
/// Cancel `tok` the moment `trigger` first holds. One watcher for every guard
/// below (LAW 5) — the guards differ only in the moment they choose.
///
/// Gives up after `budget` so a test can never hang; `Fired::real` says which
/// happened. Snapshots the output directory at the instant of the cancel, so a
/// guard can assert on the WORST MOMENT it actually hit rather than the one it
/// hoped for.
fn cancel_when(
tok: &CancelToken,
dir: PathBuf,
budget: Duration,
trigger: impl Fn(&Path) -> bool + Send + 'static,
) -> std::thread::JoinHandle<Fired> {
let tok = tok.clone();
std::thread::spawn(move || {
let t0 = Instant::now();
let mut real = false;
while t0.elapsed() < budget {
if trigger(&dir) {
real = true;
break;
}
std::thread::sleep(Duration::from_millis(1));
}
let on_disk = parquet_left(&dir);
tok.cancel();
Fired {
real,
at: Instant::now(),
on_disk,
}
})
}
/// Trigger: `name` holds at least `min_bytes` — a parquet writer has committed
/// row groups, so an unguarded interruption strands a footerless file.
fn once_written(name: &'static str, min_bytes: u64) -> impl Fn(&Path) -> bool + Send + 'static {
move |dir| {
std::fs::metadata(dir.join(name))
.map(|m| m.len())
.unwrap_or(0)
>= min_bytes
}
}
/// Trigger: the phase log records that `phase` has been entered.
fn once_phase_started(phase: &'static str) -> impl Fn(&Path) -> bool + Send + 'static {
move |dir| phase_started(&phase_log(dir), phase)
}
/// **Baseline — the fixture really does produce a complete extract.**
///
/// Without this the cancellation guards below could pass on a converter that
/// writes nothing at all: "no parquet on disk" is the same observation for
/// "cleaned up correctly" and "never worked".
#[test]
fn an_uncancelled_run_leaves_three_readable_parquet_files() {
let dir = tempfile::tempdir().unwrap();
let t0 = Instant::now();
osm_katana::convert(&fixture_bz2(), &opts(dir.path(), 4, None)).expect("convert succeeds");
let wall = t0.elapsed();
let left = parquet_left(dir.path());
assert_eq!(
left.len(),
3,
"a completed convert leaves all three products: {left:?}"
);
for (name, _) in &left {
let c = osm_katana::verify::verify(&dir.path().join(name))
.unwrap_or_else(|e| panic!("{name} must have a readable footer: {e}"));
assert!(c.rows > 0, "{name} must hold rows, got {}", c.rows);
}
let log = phase_log(dir.path());
assert!(
phase_started(&log, "pass1") && phase_started(&log, "pass2"),
"a complete run runs both passes"
);
assert!(
log.contains("\"ok\":\"true\""),
"the convert phase closes ok"
);
eprintln!(
"[fixture] full liechtenstein convert: {:.2}s",
wall.as_secs_f64()
);
}
/// **A cancel before the first work unit opens no writer at all.**
#[test]
fn cancel_before_start_opens_nothing() {
let dir = tempfile::tempdir().unwrap();
let tok = CancelToken::new();
tok.cancel();
let err = osm_katana::convert(&fixture_bz2(), &opts(dir.path(), CANCEL_WORKERS, Some(tok)))
.expect_err("a pre-cancelled convert must not run");
assert!(
is_cancelled_error(&err),
"must fail AS a cancellation: {err:#}"
);
assert_eq!(
parquet_left(dir.path()),
Vec::new(),
"no output file may exist"
);
assert!(
!phase_started(&phase_log(dir.path()), "pass1"),
"pass 1 never started"
);
}
/// **THE guard: asked to stop in pass 1, it stops — and pass 2 never begins.**
///
/// Triggered off the phase log the instant pass 1 is entered, so the phase this
/// asserts is not a race: the fixture's pass 1 runs for ~1.1 s at
/// `CANCEL_WORKERS`, and the cancel lands within milliseconds of its start.
///
/// `pass2` never starting is the assertion that fails when the signal is
/// ignored — it is the difference between "stopped" and "ran to completion".
#[test]
fn cancel_in_pass1_stops_the_run_and_pass2_never_begins() {
let dir = tempfile::tempdir().unwrap();
let tok = CancelToken::new();
let watcher = cancel_when(
&tok,
dir.path().to_path_buf(),
Duration::from_secs(120),
once_phase_started("pass1"),
);
let err = osm_katana::convert(
&fixture_bz2(),
&opts(dir.path(), CANCEL_WORKERS, Some(tok.clone())),
)
.expect_err("a cancelled convert must fail, not silently finish");
let returned = Instant::now();
let fired = watcher.join().unwrap();
assert!(fired.real, "the watcher must have seen pass 1 start");
assert!(
is_cancelled_error(&err),
"must fail AS a cancellation: {err:#}"
);
let site = tok
.stop_site()
.expect("a checkpoint must have recorded the stop");
assert_eq!(site.phase, "pass1", "stopped inside pass 1, got {site:?}");
let log = phase_log(dir.path());
assert!(phase_started(&log, "pass1"), "pass 1 really ran");
assert!(
!phase_started(&log, "pass2"),
"a convert cancelled in pass 1 must never enter pass 2 — THIS is the \
assertion that fails if the signal is ignored and the run completes:\n{log}"
);
assert!(
log.contains("\"ok\":\"false\"") && log.contains("cancelled during pass1"),
"the convert phase must close with the cancellation as its reason:\n{log}"
);
assert_eq!(
parquet_left(dir.path()),
Vec::new(),
"no parquet may survive"
);
// MEASURED, fixture: wall from the cancel request to `convert` returning.
eprintln!(
"[fixture] pass1: stopped {:.3}s after the cancel request, at work unit {}",
returned.saturating_duration_since(fired.at).as_secs_f64(),
site.units_done,
);
}
/// **The worst moment, asserted rather than hoped for: real row groups are on
/// disk when the cancel lands, and none of them survive it.**
///
/// The trigger is `nodes.parquet` reaching real bytes, and the watcher records
/// what was on disk at that instant — so the guard proves it hit a half-written
/// file rather than merely assuming it did. It deliberately makes NO claim about
/// which phase caught the signal; that is what the guard above pins.
#[test]
fn cancel_with_row_groups_already_on_disk_leaves_no_parquet() {
let dir = tempfile::tempdir().unwrap();
let tok = CancelToken::new();
let watcher = cancel_when(
&tok,
dir.path().to_path_buf(),
Duration::from_secs(120),
once_written("nodes.parquet", 4096),
);
let err = osm_katana::convert(
&fixture_bz2(),
&opts(dir.path(), CANCEL_WORKERS, Some(tok.clone())),
)
.expect_err("a cancelled convert must fail");
let fired = watcher.join().unwrap();
assert!(fired.real, "the watcher must have seen real parquet bytes");
assert!(
fired
.on_disk
.iter()
.any(|(n, b)| n == "nodes.parquet" && *b >= 4096),
"the cancel must have landed with row groups already written: {:?}",
fired.on_disk,
);
assert!(
is_cancelled_error(&err),
"must fail AS a cancellation: {err:#}"
);
assert_eq!(
parquet_left(dir.path()),
Vec::new(),
"a cancelled convert must leave NO parquet — a footerless nodes.parquet \
under a complete file's name is exactly the artefact that must not exist; \
{:?} was on disk when the cancel landed",
fired.on_disk,
);
assert!(
!dir.path().join("node_coords.arrow").exists()
&& !dir.path().join("node_coords.bin").exists(),
"the node-coord intermediates go too"
);
}
/// **Same guard, one phase later:** cancelled inside pass 2, after pass 1 has
/// already produced a *complete* `nodes.parquet`. The half-written product is
/// then `ways.parquet` — and the finished `nodes.parquet` must be removed with
/// it, because a directory holding nodes-but-no-ways is precisely the extract
/// that looks complete and is not.
#[test]
fn cancel_mid_pass2_removes_even_the_parquet_that_was_already_finished() {
let dir = tempfile::tempdir().unwrap();
let tok = CancelToken::new();
let watcher = cancel_when(
&tok,
dir.path().to_path_buf(),
Duration::from_secs(180),
// Triggered off the phase log, not off `ways.parquet` growing: pass 2 is
// the LAST phase, so a trigger that can fire late is a trigger that can
// fire after the convert has already finished. Entering pass 2 is the
// deterministic moment, and at it `nodes.parquet` is already complete —
// which is the artefact this guard is about.
once_phase_started("pass2"),
);
let err = osm_katana::convert(
&fixture_bz2(),
&opts(dir.path(), CANCEL_WORKERS, Some(tok.clone())),
)
.expect_err("a cancelled convert must fail");
let returned = Instant::now();
let fired = watcher.join().unwrap();
assert!(fired.real, "the watcher must have seen pass 2 start");
// The worst moment, asserted: a COMPLETE nodes.parquet is on disk and the
// rest of the extract is not — the state a naive interruption ships.
let nodes_at_cancel = fired
.on_disk
.iter()
.find(|(n, _)| n == "nodes.parquet")
.map(|(_, b)| *b);
assert!(
nodes_at_cancel.is_some_and(|b| b > 1_000_000),
"pass 1 must already have produced a complete nodes.parquet: {:?}",
fired.on_disk,
);
eprintln!(
"[fixture] pass2: stopped {:.3}s after the cancel request; on disk at that \
instant: {:?}",
returned.saturating_duration_since(fired.at).as_secs_f64(),
fired.on_disk,
);
assert!(
is_cancelled_error(&err),
"must fail AS a cancellation: {err:#}"
);
let site = tok.stop_site().expect("a checkpoint recorded the stop");
assert_eq!(site.phase, "pass2", "stopped inside pass 2, got {site:?}");
let log = phase_log(dir.path());
assert!(phase_started(&log, "pass2"), "pass 2 really ran");
assert!(
log.contains("\"ok\":\"false\"") && log.contains("cancelled during pass2"),
"the convert phase closes with the cancellation as its reason:\n{log}"
);
assert_eq!(
parquet_left(dir.path()),
Vec::new(),
"nodes.parquet was COMPLETE and must still go — a nodes-only directory is \
a half-written extract that looks whole"
);
}
/// **A live token that never fires changes nothing.**
///
/// The checkpoint is one relaxed atomic load per work unit; this pins that the
/// convert it guards produces the identical product, so the seam cannot be
/// paying for itself in correctness.
#[test]
fn a_live_but_unfired_token_produces_the_identical_extract() {
let plain = tempfile::tempdir().unwrap();
let guarded = tempfile::tempdir().unwrap();
osm_katana::convert(&fixture_bz2(), &opts(plain.path(), 4, None)).unwrap();
osm_katana::convert(
&fixture_bz2(),
&opts(guarded.path(), 4, Some(CancelToken::new())),
)
.unwrap();
for name in PARQUET {
let a = osm_katana::verify::verify(&plain.path().join(name)).unwrap();
let b = osm_katana::verify::verify(&guarded.path().join(name)).unwrap();
assert_eq!(a.rows, b.rows, "{name}: row count must not move");
assert_eq!(
a.geom_set, b.geom_set,
"{name}: geometry count must not move"
);
assert_eq!(
a.refs_total, b.refs_total,
"{name}: node-ref count must not move"
);
}
}