osm-katana/tests/nodes_carry_the_band.rs
//! **The POINT table gets what its siblings have — the band, and nothing it cannot use.**
//!
//! `ways.parquet` and `relations.resolved.parquet` have carried `min_zoom` and the
//! band-first `(min_zoom, hilbert)` write order since DELTA 3. `nodes.parquet` on the
//! live Sweden cut did not, and the consequence was measured by korp's parquet mirror on
//! 2026-09-02: a country viewport admitted 44–74 % of its 12 633 row groups and cost
//! **587–990 MB**, against 7.8–15.2 MB for the two banded tables. The mirror excluded the
//! points table rather than the fault.
//!
//! Every assertion here reads a parquet file back off disk or off its footer. Nothing
//! compares the packer with a copy of the packer.
//!
//! | test | the defect it is red for |
//! |---|---|
//! | [`a_point_table_is_banded_and_carries_no_rung_columns`] | points packed without a band, or given three byte-copies of their own geometry |
//! | [`a_country_camera_opens_few_groups_of_a_banded_point_table`] | the band written but not ORDERED on — the prune stays dead |
//! | [`verify_faults_a_packed_table_that_carries_no_band`] | the live nodes.parquet shape ships and nothing says so |
//! | [`the_dictionary_is_off_the_near_unique_columns`] | `id` and the WKB paying for a dictionary of near-unique values |
//! | [`an_untagged_node_lands_on_the_rung_the_ladder_names`] | the 99 % case decided by accident instead of by the ladder |
//! | [`a_second_pack_of_a_packed_points_table_is_already_complete`] | a pipeline re-run re-cutting an identical 1 GB file |
//!
//! Run: `cargo test -p osm-katana --test nodes_carry_the_band -- --nocapture`
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use arrow::array::{Array, BinaryArray, Int32Array, Int64Array, RecordBatch, StringArray, UInt8Array};
use arrow::datatypes::{DataType, Field, Schema};
use osm_katana::metadata::{self, GeoMeta};
use osm_katana::shared::class::{RUNGS, min_zoom_of_tags};
use osm_katana::shared::schema::{GENERALISED_RUNGS, generalised_geometry_name};
use osm_katana::spatial::{PackOptions, is_band_first, is_packed, pack_in_place, spatial_pack};
use osm_katana::verify::{Fault, verify};
use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::file::statistics::Statistics;
/// The tag mix of a REAL points table, and the shares are the live cut's shape: the
/// overwhelming majority of `nodes.parquet` rows are untagged way-member vertices that
/// carry `{}`, and the tagged minority spreads over the ladder.
///
/// The untagged share is what makes the band worth having on points at all — see
/// [`an_untagged_node_lands_on_the_rung_the_ladder_names`].
const TAGS: [(&str, usize); 5] = [
(r#"{}"#, 90),
(r#"{"name":"Stockholm","place":"city"}"#, 1),
(r#"{"name":"Härnevi","place":"village"}"#, 3),
(r#"{"amenity":"cafe","name":"Kafé"}"#, 4),
(r#"{"natural":"coastline"}"#, 2),
];
const N: usize = 65_536;
/// Fine groups, so the prune has enough of them to be a ratio rather than an anecdote.
const RG: usize = 1_024;
fn scratch(stem: &str) -> PathBuf {
let d = PathBuf::from(std::env::var_os("HOME").unwrap_or_default())
.join("scratch/nd-lane-test")
.join(format!("nodes-band-{}", std::process::id()));
std::fs::create_dir_all(&d).unwrap();
d.join(format!("{stem}.parquet"))
}
fn splitmix(mut x: u64) -> u64 {
x = x.wrapping_add(0x9E37_79B9_7F4A_7C15);
let mut z = x;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
/// A deterministic point over a Sweden-sized box, spatially UNCORRELATED with the row
/// index — a café and a way-member vertex sit in the same square kilometre, exactly as
/// they do in the extract.
fn scatter(i: u64) -> (f64, f64) {
let h = splitmix(i);
let x = (h & 0xFFFF_FFFF) as f64 / 4_294_967_296.0;
let y = (h >> 32) as f64 / 4_294_967_296.0;
(11.0 + x * 13.0, 55.0 + y * 14.0)
}
/// WKB Point, little-endian — the shape `writer::write_nodes` emits.
fn wkb_point(x: f64, y: f64) -> Vec<u8> {
let mut v = vec![1u8];
v.extend_from_slice(&1u32.to_le_bytes());
v.extend_from_slice(&x.to_le_bytes());
v.extend_from_slice(&y.to_le_bytes());
v
}
fn tags_of(i: usize) -> &'static str {
let r = (splitmix(i as u64 ^ 0x5EED) % 100) as usize;
let mut acc = 0;
for (t, pct) in TAGS {
acc += pct;
if r < acc {
return t;
}
}
TAGS[0].0
}
/// **The PRE-DELTA-1 points table** — `id, geometry, tags, version`, no `min_zoom`, no
/// covering. This is byte-for-byte the schema of
/// `/home/korp/.korp/work/sweden/unpacked-backup/nodes.parquet`, and what a pack must be
/// able to upgrade.
///
/// `banded`/`covering` let the same builder produce the OTHER two shapes the guard has to
/// tell apart: an unpacked table (honestly rung-less) and the live packed-but-unbanded one.
fn write_source(path: &Path, banded: bool, covering: bool) {
let mut fields = vec![
Field::new("id", DataType::Int64, false),
Field::new("geometry", DataType::Binary, true),
Field::new("tags", DataType::Utf8, true),
Field::new("version", DataType::Int32, true),
];
if banded {
fields.push(osm_katana::shared::schema::min_zoom_field());
}
if covering {
fields.push(osm_katana::spatial::bbox_field());
}
let mut meta = std::collections::HashMap::new();
meta.insert(String::from(metadata::GEO_KEY), GeoMeta::new(&["Point"]).to_json());
let schema = Arc::new(Schema::new_with_metadata(fields, meta));
let mut ids = Vec::with_capacity(N);
let mut geoms: Vec<Vec<u8>> = Vec::with_capacity(N);
let mut tags: Vec<&str> = Vec::with_capacity(N);
let mut mz = Vec::with_capacity(N);
let (mut xs, mut ys) = (Vec::with_capacity(N), Vec::with_capacity(N));
for i in 0..N {
let (x, y) = scatter(i as u64);
ids.push(splitmix(i as u64 ^ 0xF00D) as i64 & 0x0000_000F_FFFF_FFFF);
geoms.push(wkb_point(x, y));
tags.push(tags_of(i));
mz.push(min_zoom_of_tags(tags_of(i)));
xs.push(x as f32);
ys.push(y as f32);
}
let mut cols: Vec<Arc<dyn Array>> = vec![
Arc::new(Int64Array::from(ids)),
Arc::new(BinaryArray::from_iter_values(geoms.iter().map(Vec::as_slice))),
Arc::new(StringArray::from(tags)),
Arc::new(Int32Array::from(vec![1i32; N])),
];
if banded {
cols.push(Arc::new(UInt8Array::from(mz)));
}
if covering {
let f32c = |v: &Vec<f32>| Arc::new(arrow::array::Float32Array::from(v.clone())) as Arc<dyn Array>;
cols.push(Arc::new(arrow::array::StructArray::from(vec![
(Arc::new(Field::new("xmin", DataType::Float32, true)), f32c(&xs)),
(Arc::new(Field::new("ymin", DataType::Float32, true)), f32c(&ys)),
(Arc::new(Field::new("xmax", DataType::Float32, true)), f32c(&xs)),
(Arc::new(Field::new("ymax", DataType::Float32, true)), f32c(&ys)),
])));
}
let f = std::fs::File::create(path).unwrap();
let props = parquet::file::properties::WriterProperties::builder()
.set_max_row_group_row_count(Some(RG))
.build();
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema.clone(), Some(props)).unwrap();
w.write(&RecordBatch::try_new(schema, cols).unwrap()).unwrap();
w.close().unwrap();
}
fn pack(src: &Path, dst: &Path) {
let opts = PackOptions { row_group_rows: RG, ..PackOptions::default() };
spatial_pack(src, dst, &opts).unwrap_or_else(|e| panic!("spatial_pack {src:?}: {e:#}"));
}
/// Every row group's `min_zoom` `(min, max)`, off the footer. No row is decoded.
fn bands(path: &Path) -> Vec<(u8, u8)> {
let r = SerializedFileReader::new(std::fs::File::open(path).unwrap()).unwrap();
let md = r.metadata();
(0..md.num_row_groups())
.map(|g| {
let rg = md.row_group(g);
let col = rg
.columns()
.iter()
.find(|c| c.column_path().string() == "min_zoom")
.unwrap_or_else(|| panic!("row group {g} has no min_zoom column chunk"));
match col.statistics().expect("min_zoom statistics") {
Statistics::Int32(s) => (*s.min_opt().unwrap() as u8, *s.max_opt().unwrap() as u8),
other => panic!("min_zoom statistics came back as {other:?}"),
}
})
.collect()
}
/// The groups a `min_zoom <= camera` reader must OPEN — the prune, computed the way
/// korp's `Footer::admits` computes it: a group whose SHALLOWEST rung is deeper than the
/// camera holds nothing the camera may draw.
fn opened_at(bands: &[(u8, u8)], camera: u8) -> usize {
bands.iter().filter(|(lo, _)| *lo <= camera).count()
}
// ── 1. the band, and the rung DECISION for points ───────────────────────────────
/// ★ **RED-when-broken: a packed point table with no `min_zoom`, or with rung columns.**
///
/// Two halves of one decision, both read off disk.
///
/// The band is ADDED — the source has `tags` and no `min_zoom`, exactly like the live
/// `unpacked-backup/nodes.parquet` — and every value is the one ladder's answer for that
/// row's own tags. Non-vacuity: the written rungs must SPAN more than one rung, or a
/// packer writing a constant would agree with itself everywhere.
///
/// The rung columns are ABSENT, and that is the point-table decision stated as applied
/// output rather than as a comment. A `geometry_z6` on a POINT is Douglas–Peucker applied
/// to a single vertex, i.e. a byte-for-byte copy of `geometry` — measured at **+1.97 GB**
/// on the live Sweden `nodes.parquet` (655.18 MB of compressed point geometry, three
/// times) buying a reader nothing the base column does not already give it. `min_zoom`
/// alone gates a point; the READ projects the one geometry column it needs.
#[test]
fn a_point_table_is_banded_and_carries_no_rung_columns() {
let src = scratch("src-points");
let dst = scratch("dst-points");
write_source(&src, false, false);
pack(&src, &dst);
let f = std::fs::File::open(&dst).unwrap();
let rdr = parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(f)
.unwrap()
.build()
.unwrap();
let mut per_rung: BTreeMap<u8, usize> = BTreeMap::new();
let mut rows = 0usize;
let mut wrong = 0usize;
for b in rdr {
let b = b.unwrap();
let tags = b.column_by_name("tags").unwrap().as_any().downcast_ref::<StringArray>().unwrap();
let mz = b
.column_by_name("min_zoom")
.expect("the packed POINT table carries no `min_zoom` column — the band never \
reached the table that is 12 633 of the cut's 13 806 row groups")
.as_any()
.downcast_ref::<UInt8Array>()
.expect("min_zoom is UInt8");
// The rung columns are absent from the SCHEMA, which is the strong form: not
// "present and null", not "present and equal to geometry" — not there at all.
for &z in &GENERALISED_RUNGS {
let name = generalised_geometry_name(z);
assert!(
b.column_by_name(&name).is_none(),
"the point table carries `{name}` — a point has no vertex to drop, so \
that column is a byte-for-byte copy of `geometry` (+1.97 GB on the live \
Sweden cut) and carries no information"
);
}
for i in 0..b.num_rows() {
assert!(!mz.is_null(i), "min_zoom is NULL on a row — the column must be NOT NULL");
if mz.value(i) != min_zoom_of_tags(tags.value(i)) {
wrong += 1;
}
*per_rung.entry(mz.value(i)).or_default() += 1;
rows += 1;
}
}
assert_eq!(rows, N, "the pack dropped rows: {rows} of {N}");
assert_eq!(
wrong, 0,
"{wrong} of {N} point rows carry a min_zoom that is not what the ONE ladder gives \
their own tags"
);
assert!(
per_rung.len() >= 3,
"the written rungs {per_rung:?} do not spread — a constant column would pass the \
row-by-row check above"
);
for r in per_rung.keys() {
assert!(RUNGS.contains(r), "rung {r} is not on the ladder {RUNGS:?}");
}
println!("gate 1: {rows} points, rows per rung {per_rung:?}, no rung columns");
}
// ── 2. the ORDER, and what it buys a country camera ─────────────────────────────
/// ★ **RED-when-broken: the band is written but not ordered on, so nothing prunes.**
///
/// Off the footer only. Band-first means `max(g) <= min(g + 1)` for consecutive groups —
/// [`is_band_first`], the packer's own predicate, so this cannot drift from the writer —
/// and the consequence is that a `min_zoom <= 6` reader opens a small FRACTION of the
/// file.
///
/// **The RED arm is a real file, not a mutation of this one**: the same rows written in
/// input order with the band column present and no band ordering. Every group there
/// straddles the ladder, so a country camera opens all of them — which is the live
/// `nodes.parquet` behaviour this lane exists to end. The test fails if the two arms are
/// not far apart, so a packer that stopped sorting cannot pass it.
#[test]
fn a_country_camera_opens_few_groups_of_a_banded_point_table() {
let src = scratch("src-order");
let dst = scratch("dst-order");
write_source(&src, false, false);
pack(&src, &dst);
let packed = bands(&dst);
let total = packed.len();
assert!(total >= 32, "only {total} row groups — too few to measure a prune");
let straddle: Vec<usize> = packed
.windows(2)
.enumerate()
.filter(|(_, w)| w[0].1 > w[1].0)
.map(|(g, _)| g)
.collect();
assert!(
straddle.is_empty(),
"{} of {total} consecutive row-group pairs are NOT monotone in min_zoom (first at \
{:?}) — the write order is not band-first",
straddle.len(),
straddle.first()
);
let md = SerializedFileReader::new(std::fs::File::open(&dst).unwrap()).unwrap();
assert!(
is_band_first(md.metadata()),
"the packer's own band-first predicate refuses the file the packer just wrote"
);
// The RED arm: the same rows, the band column present, NO band ordering.
let bare = scratch("src-unordered");
write_source(&bare, true, false);
let unordered = bands(&bare);
// The CAMERAS the prune is for. 90 % of a points table is untagged vertices on rung
// 12, so a z12 camera legitimately opens (nearly) everything — the win is at the
// country and regional cameras, where the whole rung-12 mass sits behind one integer
// compare. Asserting a prune at 12 would be asserting that data goes missing.
let mut report = Vec::new();
for camera in [6u8, 9] {
let hot = opened_at(&packed, camera);
let cold = opened_at(&unordered, camera);
report.push(format!("z{camera}: {hot}/{total} banded vs {cold}/{} unordered", unordered.len()));
assert!(
hot * 4 < cold,
"at camera {camera} the banded file opens {hot} of {total} groups and the \
unordered one {cold} of {} — the band is written but the ORDER is not on it, \
so the prune is dead",
unordered.len()
);
}
// Non-vacuity at the other end: a camera at or below every rung the file holds must
// open EVERY group, or the "prune" above is just a file that lost rows.
let deepest = packed.iter().map(|(_, hi)| *hi).max().unwrap();
assert_eq!(
opened_at(&packed, deepest),
total,
"a camera at the deepest rung in the file must still open every group — the prune \
dropped data"
);
report.push(format!("z{deepest}: {total}/{total} (all of it, as it must be)"));
println!("gate 2: {}", report.join(" · "));
}
// ── 3. the guard that would have caught the live file ───────────────────────────
/// ★ **RED-when-broken: `verify` is silent about a packed table with no band.**
///
/// Three shapes, three verdicts, and the middle one is the finding:
///
/// * PACKED + `tags` + banded ⇒ no fault. (If this faulted, the guard would fail a
/// healthy cut, which is the shape LAW 2 exists to prevent.)
/// * PACKED + `tags` + NOT banded ⇒ [`Fault::UnbandedPacked`]. This is
/// `~/.korp/work/sweden/nodes.parquet` as it stood on 2026-09-03.
/// * UNPACKED + `tags` + NOT banded ⇒ no fault. A convert output on its way to the pack
/// pass is not a defect, and a guard that could not tell the two apart would fault
/// every intermediate in the pipeline.
#[test]
fn verify_faults_a_packed_table_that_carries_no_band() {
let good = scratch("verdict-banded");
write_source(&good, false, false);
let packed_good = scratch("verdict-banded-packed");
pack(&good, &packed_good);
let v = verify(&packed_good).unwrap();
assert!(v.packed && v.banded && v.taggable, "the healthy arm is not the shape it claims: packed={} banded={} taggable={}", v.packed, v.banded, v.taggable);
assert_eq!(
v.fault(true),
None,
"a packed, banded points table was faulted — the guard fails a healthy cut"
);
// THE LIVE SHAPE: `covering.bbox` and `tags` present, `min_zoom` absent.
let bad = scratch("verdict-unbanded-packed");
write_source(&bad, false, true);
let v = verify(&bad).unwrap();
assert!(v.packed, "the RED arm has no covering column, so it is not the live shape");
assert!(!v.banded, "the RED arm carries a band, so it is not the live shape");
match v.fault(true) {
Some(Fault::UnbandedPacked { rows, row_groups }) => {
assert_eq!(rows, N);
assert!(row_groups > 1, "one row group cannot demonstrate a prune");
println!("gate 3: {}", Fault::UnbandedPacked { rows, row_groups });
}
other => panic!(
"a PACKED points table with `tags` and no `min_zoom` verified as {other:?} — \
this is the exact file korp's mirror measured at 587-990 MB per country \
viewport, and `verify` said nothing about it"
),
}
// The control: an UNPACKED table with no band is an honest intermediate.
let raw = scratch("verdict-unpacked");
write_source(&raw, false, false);
let v = verify(&raw).unwrap();
assert!(!v.packed && !v.banded);
assert_eq!(
v.fault(true),
None,
"convert's own output was faulted for lacking a band it has not been packed with \
yet — the guard cannot tell a pipeline stage from a defect"
);
}
// ── 4. the two fat columns ──────────────────────────────────────────────────────
/// ★ **RED-when-broken: `id` and the WKB pay for a dictionary of near-unique values.**
///
/// APPLIED OUTPUT, read off the footer of a file the packer wrote: the `id` and
/// `geometry` column chunks must carry NO dictionary page, and `tags` — whose values
/// repeat — must still carry one. A test that only asserted "no dictionary anywhere"
/// would pass on a writer that had turned the dictionary off everywhere, which is a
/// different and worse file.
///
/// The measurement that chose this is in [`osm_katana::spatial`]'s `dictionary_is_waste`:
/// on the live `nodes.parquet`, `id` 1 535 907 → 748 870 bytes and `geometry`
/// 3 118 459 → 2 328 614 over a 60-group sample, i.e. **−24.2 % of the whole file**.
#[test]
fn the_dictionary_is_off_the_near_unique_columns() {
let src = scratch("src-dict");
let dst = scratch("dst-dict");
write_source(&src, false, false);
pack(&src, &dst);
let r = SerializedFileReader::new(std::fs::File::open(&dst).unwrap()).unwrap();
let md = r.metadata();
let mut dict: BTreeMap<String, (usize, usize)> = BTreeMap::new();
for g in 0..md.num_row_groups() {
for c in md.row_group(g).columns() {
let e = dict.entry(c.column_path().string()).or_insert((0, 0));
e.1 += 1;
if c.dictionary_page_offset().is_some() {
e.0 += 1;
}
}
}
for near_unique in ["id", "geometry"] {
let (with, all) = dict[near_unique];
assert_eq!(
with, 0,
"{with} of {all} `{near_unique}` chunks still carry a dictionary page — the \
values are near-unique, so the dictionary is a second copy of the column \
with an index array bolted on"
);
}
let (with, all) = dict["tags"];
assert_eq!(
with, all,
"only {with} of {all} `tags` chunks carry a dictionary — the tag JSON REPEATS and \
the dictionary is the whole reason that column is 1.9 % of the file; this change \
must not have turned it off everywhere"
);
println!("gate 4: dictionary pages per column {dict:?}");
}
// ── 5. the 99 % case ────────────────────────────────────────────────────────────
/// **What rung an UNTAGGED node lands on, and why it is a decision.**
///
/// The overwhelming majority of `nodes.parquet` rows are way-member vertices with `{}`
/// for tags: they exist so a way can reference them, and they carry no feature. The
/// ladder matches no rule for them, so they take `UNCLASSIFIED_MIN_ZOOM` — the City rung,
/// **12** — which is `classify_with`'s own fallthrough on the painting side. That is what
/// makes the band worth having on points: at a country or regional camera every one of
/// those rows is behind a rung the camera does not open, so the row groups holding them
/// are never touched.
///
/// The arms are literals. A test that recomputed `min_zoom_of_tags` and compared it with
/// itself would be green on any ladder.
#[test]
fn an_untagged_node_lands_on_the_rung_the_ladder_names() {
assert_eq!(
min_zoom_of_tags("{}"),
12,
"an untagged node must land on the City rung — coarser and it paints over the \
road network at a country camera (the `osm-other` defect); deeper and a z12 \
camera loses the vertices it is entitled to"
);
assert_eq!(min_zoom_of_tags(""), 12, "an EMPTY tags cell is the same case as `{{}}`");
// Non-vacuity: the ladder must not answer 12 to everything.
assert_eq!(min_zoom_of_tags(r#"{"place":"city","name":"Stockholm"}"#), 6);
assert_eq!(min_zoom_of_tags(r#"{"place":"village"}"#), 9);
assert_eq!(min_zoom_of_tags(r#"{"highway":"path"}"#), 15);
// …and the untagged answer is not simply the ladder's floor or ceiling.
assert!(
min_zoom_of_tags("{}") > RUNGS[0] && min_zoom_of_tags("{}") < RUNGS[3],
"the untagged rung is at an END of the ladder {RUNGS:?} — that is a default, not \
a decision"
);
}
// ── 6. a re-run is a no-op, not a second cut ────────────────────────────────────
/// ★ **RED-when-broken: packing a packed points table cuts it again.**
///
/// The pipeline's own idempotence, on the POINTS shape specifically. [`is_packed`] used to
/// ask only for `covering.bbox`; DELTA 3 taught it to ask for the band and the ORDER too,
/// which is what makes an old unbanded file upgrade — and this is the other half of that
/// contract: once upgraded, a re-run must report **already complete** rather than spend
/// 48 s and 17 GB re-cutting an identical file.
///
/// Three arms:
///
/// * an unbanded packed points table (the live shape) ⇒ `is_packed == false`, so the
/// pipeline WILL re-cut it. Without this arm the test could pass on an `is_packed` that
/// always says `true`.
/// * the packer's own output ⇒ `is_packed == true` and [`pack_in_place`] returns `None`.
/// * `$KATANA_PACKED_NODES`, when set, gets the same two questions asked of a REAL file —
/// so the artefact a lane hands over can be checked without a fixture standing in for it.
#[test]
fn a_second_pack_of_a_packed_points_table_is_already_complete() {
let src = scratch("src-idem");
let dst = scratch("dst-idem");
write_source(&src, false, false);
pack(&src, &dst);
// The live shape must NOT read as packed, or nothing would ever be upgraded.
let bad = scratch("idem-unbanded-packed");
write_source(&bad, false, true);
assert!(
!is_packed(&bad).unwrap(),
"a points table with `covering.bbox`, `tags` and NO `min_zoom` read as PACKED — \
the live nodes.parquet would then never be upgraded by any pipeline re-run"
);
assert!(
is_packed(&dst).unwrap(),
"the packer's own output does not read as packed — every re-run would re-cut it"
);
let before = std::fs::metadata(&dst).unwrap().len();
let again = pack_in_place(&dst, &PackOptions { row_group_rows: RG, ..PackOptions::default() })
.expect("pack_in_place");
assert!(
again.is_none(),
"a second pack re-cut an already-packed points table ({again:?}) — on the real \
table that is 48 s and 17 GB of peak for an identical file"
);
assert_eq!(
std::fs::metadata(&dst).unwrap().len(),
before,
"the file changed size on a re-run that claimed to do nothing"
);
// The REAL artefact, when a lane points at one.
if let Ok(real) = std::env::var("KATANA_PACKED_NODES") {
let p = std::path::Path::new(&real);
assert!(
is_packed(p).unwrap(),
"{real} does not read as packed — a pipeline re-run would re-cut the artefact"
);
println!("gate 6: {real} reads as packed; a re-run is already complete");
}
}