src/bin/korp-chain-init.rs
//! `korp-chain-init` — **the KVM guest PID 1 that drives the OSM collector chain a–e.**
//!
//! # Why this exists
//!
//! The owner: *"a b c d e är inte bara tester, det är hur collectorn OSM skall funka"* — the
//! chain must survive a restart, keep producing artefacts, and **the moment one is
//! incomplete: nuke it and reproduce from the last healthy step.** None of those three
//! properties can be proven in a `cargo test` on the host:
//!
//! * **A restart is not a `drop()`.** Every host-side "interruption" test tears the
//! process down cooperatively. The failure mode that actually bit this house
//! (znippy-zoomies `49c7bc6`: a `SIGTERM` that logged *"stop latency 1–15 ms"* and then
//! died `status=11/SEGV` three seconds later with a detached background sorter still
//! running) needs a machine that can be **cut off at the power**, which is
//! `BootHandle::kill()` on a QEMU process — nothing softer reproduces it.
//! * **`st_blocks` is a property of a real filesystem.** The sparse-planet check
//! ([`korp_collectors::osm_cut::planet_complete`]) reads how many blocks the
//! filesystem actually allocated. On tmpfs — which is where `/tmp` lives on oden, and
//! where every host tempdir test therefore runs — those numbers do not behave like
//! ext4's. The guest mounts a **real ext4** image, so the probe is asked the question
//! it will be asked in production.
//! * **A chain that resumes must resume across a boot**, from nothing but what is on
//! disk. Two VM boots over one disk image is that statement, exactly.
//!
//! # Contract
//!
//! PID 1: mount `/proc`, `/sys`, `/dev`, then the work disk (`/dev/vda`, ext4) at
//! `/work`. Read the plan from `/proc/cmdline`. Run it. Write a JSON report to the
//! **mailbox** block device (`/dev/vdb`) as `u64-LE length || JSON`, echo it on the
//! serial console, `sync`, and power off.
//!
//! The mailbox is a raw device with no filesystem on purpose: the host reads the report
//! back with a `File::read` and no loop mount, no root, and no `mkfs` round-trip that
//! could itself be the thing that fails.
//!
//! **Every marker this prints is derived from a file that was read back off the disk**,
//! never from a variable the run just set. `assert!(r.drew())` is blind; so is
//! `assert!(cut_ok)`.
//!
//! # Running it
//!
//! Never by hand — [`korp-collectors/tests/osm_chain_kvm.rs`] builds it static, packs it
//! into an initramfs with the Liechtenstein fixture, and boots it. See that file for the
//! cell list.
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::time::Instant;
use korp_collectors::daemon::LogSink;
use korp_collectors::osm_cut;
/// The work disk (ext4, formatted by the host) — where the extract and the tile tree
/// live, and the ONLY thing that survives between the two boots of the restart cell.
const WORK_DEV: &str = "/dev/vda";
/// The mailbox (raw, no filesystem) — where the report goes back to the host.
const MAILBOX_DEV: &str = "/dev/vdb";
/// Where the fixture rides in the initramfs.
const FIXTURE_DIR: &str = "/fixture";
fn main() {
let t0 = Instant::now();
let is_pid1 = std::process::id() == 1;
if is_pid1 {
// Nothing is mounted in an initramfs. Do this before anything reads a path.
mount_or_die("proc", "/proc", "proc");
mount_or_die("sysfs", "/sys", "sysfs");
mount_or_die("devtmpfs", "/dev", "devtmpfs");
}
println!("KORP-CHAIN BOOT-OK: guest reached runtime (pid1={is_pid1})");
let plan = Plan::from_cmdline();
println!("KORP-CHAIN plan {plan:?}");
let mut report = serde_json::Map::new();
report.insert("run".into(), plan.run.into());
report.insert("region".into(), plan.region.clone().into());
// The work disk. A failure here is a REPORTED failure, never a skip: a cell that
// cannot run must say so as its own outcome (a silent skip is how four of eight
// invariants passed vacuously in the release engine's dry run).
match mount(WORK_DEV, "/work", "ext4") {
Ok(()) => report.insert("work_mounted".into(), true.into()),
Err(e) => {
report.insert("work_mounted".into(), false.into());
report.insert(
"fatal".into(),
format!("mount {WORK_DEV} -> /work: {e}").into(),
)
}
};
if report.get("work_mounted") == Some(&serde_json::Value::Bool(true)) {
run_plan(&plan, Path::new("/work"), &mut report);
}
report.insert("wall_ms".into(), u64_json(t0.elapsed().as_millis() as u64));
finish(report, is_pid1);
}
// ─────────────────────────────────────────────────────────────────────────────
// The plan — read off the kernel cmdline, because that is the one channel a
// firmware boot has before any filesystem exists.
// ─────────────────────────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
struct Plan {
/// Which boot this is (1 = first, 2 = the resume after the power cut).
run: u64,
region: String,
/// Seconds into the cut at which to die HARD (`_exit`, no unwinding, no flush) —
/// the in-guest stand-in for a power cut when the host is not driving the kill.
hard_exit_after_s: Option<u64>,
/// Run the step-a sparse-planet probe (needs a real filesystem to mean anything).
probe_sparse: bool,
/// Run the step-a torrent-source probe (resolve a source, download nothing).
probe_torrent: bool,
/// Corrupt this artefact before the run, to prove the chain detects and nukes it.
damage: Option<String>,
/// Cut under a DIFFERENT declaration than run 1, to prove a recipe change re-cuts.
recipe_bump: bool,
}
impl Plan {
fn from_cmdline() -> Plan {
let raw = std::fs::read_to_string("/proc/cmdline").unwrap_or_default();
let get = |k: &str| -> Option<String> {
raw.split_whitespace()
.find_map(|t| t.strip_prefix(k).map(|v| v.to_string()))
};
Plan {
run: get("korp.chain.run=")
.and_then(|v| v.parse().ok())
.unwrap_or(1),
region: get("korp.chain.region=").unwrap_or_else(|| "fixture".into()),
hard_exit_after_s: get("korp.chain.hardexit=").and_then(|v| v.parse().ok()),
probe_sparse: get("korp.chain.sparse=").as_deref() == Some("1"),
probe_torrent: get("korp.chain.torrent=").as_deref() == Some("1"),
damage: get("korp.chain.damage="),
recipe_bump: get("korp.chain.recipe_bump=").as_deref() == Some("1"),
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The chain
// ─────────────────────────────────────────────────────────────────────────────
fn run_plan(plan: &Plan, work: &Path, report: &mut serde_json::Map<String, serde_json::Value>) {
let disk_dir = work.join("collector");
let _ = std::fs::create_dir_all(&disk_dir);
// ── step a: is the source COMPLETE? ─────────────────────────────────────
if plan.probe_sparse {
// The AUTHORITY first, then the corroborating filesystem heuristic — in the order
// they now matter.
probe_ledger(work, report);
probe_sparse(work, report);
}
let Some(fixture) = first_fixture() else {
report.insert("fatal".into(), "no fixture in the initramfs".into());
return;
};
// The fixture is read-only in the initramfs (RAM). Copy it onto the ext4 work disk
// so `planet_complete` and `planet_identity` are asked about a REAL file on a REAL
// filesystem — the whole reason this runs in a VM.
let planet = disk_dir.join(fixture.file_name().unwrap_or_default());
if !planet.exists() {
if let Err(e) = std::fs::copy(&fixture, &planet) {
report.insert("fatal".into(), format!("seed planet: {e}").into());
return;
}
}
report.insert("planet_bytes".into(), u64_json(len_of(&planet)));
// ── Step a, on the REAL planet, decided by the ledger. ────────────────────
//
// Before any producer has spoken, the seeded fixture is `unrecorded` — even though it
// is a whole file that every filesystem heuristic accepts. That pair of facts, side by
// side in one report, IS the change of authority.
report.insert(
"planet_standing_before".into(),
osm_cut::planet_standing(&disk_dir).kind().into(),
);
report.insert(
"planet_looks_whole_on_disk".into(),
osm_cut::planet_looks_whole_on_disk(&planet).into(),
);
// The declared-planet producer speaks: this is a test/operator handoff, and it is
// recorded like any other completion instead of being re-derived on every pass.
if let Err(e) = osm_cut::record_planet(
&disk_dir,
&planet,
&format!("guest fixture {}", planet.display()),
"korp-chain-init (guest fixture seed)",
) {
report.insert(
"fatal".into(),
format!("record the seeded planet: {e:#}").into(),
);
return;
}
let standing = osm_cut::planet_standing(&disk_dir);
report.insert("planet_standing_after".into(), standing.kind().into());
report.insert("planet_standing_reason".into(), standing.reason().into());
report.insert("planet_complete".into(), standing.is_complete().into());
// The collector's OWN resolver must find it — this is the production path, not a
// direct call to the predicate.
report.insert(
"cached_planet_found".into(),
osm_cut::cached_planet(&disk_dir).is_some().into(),
);
// AFTER the planet is on disk — the whole point of the probe is that a source
// already present short-circuits acquisition, and asking before the seed exists
// measures the empty directory instead. (It did, on the first run: the cell went
// red with `seeded_planet_short_circuits=false` over a resolver that was working
// perfectly, one line further down.)
if plan.probe_torrent {
probe_torrent(&disk_dir, report);
}
// ── the damage arm: break an artefact, then let the chain meet it. ───────
if let Some(what) = &plan.damage {
let victim = damage_target(&disk_dir, &plan.region, what);
match &victim {
Some(p) if p.exists() => {
let before = len_of(p);
truncate_to(p, before / 2);
report.insert("damaged".into(), p.display().to_string().into());
report.insert("damaged_from".into(), u64_json(before));
report.insert("damaged_to".into(), u64_json(len_of(p)));
}
_ => {
// Naming a victim that is not there is a FAILED cell, not a skip: the
// whole point is that run 1 left it behind.
report.insert(
"fatal".into(),
format!("damage target `{what}` does not exist — run 1 did not leave it")
.into(),
);
return;
}
}
}
// What the chain believes about its own artefacts BEFORE it runs — the input to
// the nuke-and-reproduce decision.
let own_extract = disk_dir.join(&plan.region);
report.insert(
"extract_complete_before".into(),
osm_cut::extract_complete(&own_extract).into(),
);
// **The LEDGER as this boot inherited it**, before this run records anything of its
// own. This is the field that says what the PREVIOUS boot managed to seal — and
// therefore the one a power-cut cell must read, since `recorded_steps` at the end of
// the report necessarily includes whatever this run went on to finish.
report.insert(
"recorded_steps_before".into(),
recorded_steps(&disk_dir, &plan.region).into(),
);
// ── the hard-exit arm: die mid-cut, with no cleanup whatsoever. ──────────
if let Some(secs) = plan.hard_exit_after_s {
std::thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_secs(secs));
eprintln!("KORP-CHAIN HARD-EXIT: cutting power mid-cut, no unwind, no flush");
// _exit, not process::exit: no atexit, no destructor, no buffered write
// reaching the disk. As close to a power cut as a process can get.
unsafe { libc::_exit(9) };
});
}
// ── steps b–c: the REAL production cut. ─────────────────────────────────
//
// It used to end in a tile pyramid published to a `DiskStore` bucket. The pyramid
// is deleted (korp map v2), so the artefact this guest produces — and the artefact
// the power-cut cells damage and the chain reproduces — is the EXTRACT itself.
let (log, lines) = LogSink::recorder();
let config = serde_json::json!({
"clip_mode": "none",
"keep_node_coords": true,
// A recipe bump is a DIFFERENT declaration; the extract stamp must notice.
"revision": if plan.recipe_bump { "bumped" } else { "" },
"keep_generations": 2,
});
let declaration = osm_cut::cut_declaration(&config, &plan.region);
report.insert("declaration".into(), declaration.clone().into());
let cut = osm_cut::RegionCut::from_config(&config, &planet, &plan.region, &own_extract, None);
let t = Instant::now();
match osm_cut::cut_region_extract(&cut, &log) {
Ok(stats) => {
report.insert("cut_ok".into(), true.into());
report.insert("cut_ms".into(), u64_json(t.elapsed().as_millis() as u64));
report.insert("reused_extract".into(), stats.reused_extract.into());
report.insert(
"extract_dir".into(),
stats.extract_dir.display().to_string().into(),
);
report.insert(
"checks".into(),
serde_json::Value::Array(stats.checks.iter().map(|c| c.line().into()).collect()),
);
}
Err(e) => {
report.insert("cut_ok".into(), false.into());
report.insert("cut_err".into(), format!("{e:#}").into());
}
}
// ── APPLIED OUTPUT: read the artefacts back OFF THE DISK. ───────────────
//
// Not `stats.tiles_written` — that is what the run believed. These are files.
let src = if own_extract.exists() {
own_extract.clone()
} else {
disk_dir.join("regions").join(&plan.region)
};
for f in [
"nodes.parquet",
"ways.parquet",
"relations.parquet",
"node_coords.arrow",
] {
let p = src.join(f);
report.insert(format!("artefact.{f}.bytes"), u64_json(len_of(&p)));
report.insert(
format!("artefact.{f}.complete"),
if f.ends_with(".arrow") {
arrow_complete(&p).into()
} else {
osm_cut::parquet_complete(&p).into()
},
);
}
report.insert(
"extract_complete_after".into(),
osm_cut::extract_complete(&src).into(),
);
// **THE SERVING READ PATH, timed.** This is what korp-server does per map query
// now that the pyramid is gone: read the packed extract's own parquet, check its
// footer. Measured over every table the cut produced, so the number is a
// throughput and not one lucky object.
let mut read_bytes = 0u64;
let mut footerless = 0u64;
let mut tables = 0u64;
let t_read = Instant::now();
for f in ["nodes.parquet", "ways.parquet", "relations.parquet"] {
let p = src.join(f);
let Ok(bytes) = std::fs::read(&p) else {
continue;
};
tables += 1;
read_bytes += bytes.len() as u64;
if bytes.len() < 4 || &bytes[bytes.len() - 4..] != b"PAR1" {
footerless += 1;
}
}
report.insert(
"readback_ms".into(),
u64_json(t_read.elapsed().as_millis() as u64),
);
report.insert(
"readback_us".into(),
u64_json(t_read.elapsed().as_micros() as u64),
);
report.insert("readback_bytes".into(), u64_json(read_bytes));
report.insert("tables_read".into(), u64_json(tables));
report.insert("tables_footerless".into(), u64_json(footerless));
// RED-PROOF SWITCH for the b–e ledger assertion. `korp.chain.redproof_norecord=1`
// FORGETS the records the chain just sealed, reproducing a collector that produced
// artefacts and vouched for none of them — the state the whole ledger replaced. It
// exists so the b–e assertion can be SEEN failing, on demand, without unwiring the
// producer. Off unless explicitly asked for.
if std::fs::read_to_string("/proc/cmdline")
.unwrap_or_default()
.contains("korp.chain.redproof_norecord=1")
{
eprintln!("KORP-CHAIN REDPROOF: forgetting the b–e completion records");
if let Ok(store) = korp_collectors::state::StateStore::open(&disk_dir) {
let _ = store.invalidate_from(korp_collectors::state::Step::Coords, &plan.region);
}
}
// ── THE LEDGER, end to end — what the chain RECORDED, per step. ──────────
//
// This is what makes "nuke what is incomplete and reproduce from the last healthy
// step" expressible at all: the healthy steps are the ones with a record, read back
// through a lock-free reader exactly as korp's provisioner would read them.
match korp_collectors::state::StateReader::open(&disk_dir) {
Ok(Some(r)) => {
let mut recorded = serde_json::Map::new();
for step in korp_collectors::state::Step::ALL {
let key = if step == korp_collectors::state::Step::Planet {
osm_cut::PLANET_KEY
} else {
&plan.region
};
match r.get(step, key) {
Ok(Some(c)) => {
recorded.insert(
step.letter().to_string(),
serde_json::json!({
"artefact": c.artefact,
"bytes": c.bytes,
"source": c.source,
"producer": c.producer,
"declaration": c.declaration,
}),
);
}
Ok(None) => {}
Err(e) => {
recorded.insert(step.letter().to_string(), format!("ERR {e:#}").into());
}
}
}
report.insert(
"recorded_steps".into(),
serde_json::Value::String(recorded.keys().cloned().collect::<Vec<_>>().join("")),
);
report.insert("ledger".into(), serde_json::Value::Object(recorded));
}
Ok(None) => {
report.insert(
"recorded_steps".into(),
serde_json::Value::String(String::new()),
);
}
Err(e) => {
report.insert("ledger_err".into(), format!("{e:#}").into());
}
}
let l = lines.lock().map(|v| v.clone()).unwrap_or_default();
report.insert("log_lines".into(), u64_json(l.len() as u64));
report.insert(
"log".into(),
serde_json::Value::Array(l.iter().map(|s| s.clone().into()).collect()),
);
}
/// Step a, the sparse arm — **on the real ext4 work disk, not tmpfs.**
///
/// Three files, because a check that only ever meets one of them proves nothing about
/// the other two, and the middle one is where the identity-value trap lives:
/// a whole file (must read COMPLETE), a preallocated-but-empty file of the SAME
/// `st_size` (must read INCOMPLETE — this is the 1.97 %-filled planet), and a
/// zero-length file (must read INCOMPLETE, not "trivially complete").
fn probe_sparse(work: &Path, report: &mut serde_json::Map<String, serde_json::Value>) {
let d = work.join("sparse-probe");
let _ = std::fs::create_dir_all(&d);
let size = 64u64 * 1024 * 1024;
let whole = d.join("whole.osm.bz2");
if let Ok(mut f) = std::fs::File::create(&whole) {
let buf = vec![0x5au8; 1 << 20];
for _ in 0..(size / (1 << 20)) {
let _ = f.write_all(&buf);
}
let _ = f.sync_all();
}
// Exactly what librqbit does before it has a single piece: set the length, write
// nothing. `st_size` is full; `st_blocks` is ~0.
let prealloc = d.join("preallocated.osm.bz2");
if let Ok(f) = std::fs::File::create(&prealloc) {
let _ = f.set_len(size);
let _ = f.sync_all();
}
let empty = d.join("empty.osm.bz2");
let _ = std::fs::File::create(&empty);
for (name, p) in [
("whole", &whole),
("preallocated", &prealloc),
("empty", &empty),
] {
let md = std::fs::metadata(p).ok();
let on_disk = md.as_ref().map(osm_cut::bytes_on_disk).unwrap_or(0);
let sz = md.as_ref().map(|m| m.len()).unwrap_or(0);
report.insert(format!("sparse.{name}.size"), u64_json(sz));
report.insert(format!("sparse.{name}.on_disk"), u64_json(on_disk));
report.insert(
format!("sparse.{name}.complete"),
osm_cut::planet_looks_whole_on_disk(p).into(),
);
}
// And the production resolver over a directory holding ONLY the preallocated one:
// it must decline, not hand the collector a file full of holes.
let only = work.join("sparse-only");
let _ = std::fs::create_dir_all(&only);
if let Ok(f) = std::fs::File::create(only.join("planet-latest.osm.bz2")) {
let _ = f.set_len(size);
let _ = f.sync_all();
}
// RED-PROOF SWITCH. With `KORP_CHAIN_REDPROOF_SPARSE=1` on the KERNEL CMDLINE the
// guest asks the OLDEST question — "is there a file whose NAME looks like a planet?" —
// which is exactly what `cached_planet` did before 2026-08-05. It exists so the cell
// can be SEEN failing against the defect it was written for, on demand, without a
// source revert. It is off unless explicitly asked for.
let redproof = std::fs::read_to_string("/proc/cmdline")
.unwrap_or_default()
.contains("korp.chain.redproof_sparse=1");
let declines = if redproof {
eprintln!("KORP-CHAIN REDPROOF: asking the pre-fix question (filename only)");
std::fs::read_dir(&only)
.map(|rd| {
!rd.flatten().any(|e| {
let n = e.path().to_string_lossy().to_lowercase();
n.ends_with(".osm.bz2") || n.ends_with(".osm.pbf") || n.ends_with(".pbf")
})
})
.unwrap_or(true)
} else {
osm_cut::cached_planet(&only).is_none()
};
report.insert("sparse.resolver_declines".into(), declines.into());
}
/// **Step a, the LEDGER arm — the authority.**
///
/// The sparse arm above is the second opinion; this is the one that decides. Every case
/// runs on a real ext4 disk inside the guest against the real
/// [`korp_collectors::state`] store, because the redb file, its `flock` and the fsync
/// ordering are properties of a filesystem and a process — not of a tempdir in a unit
/// test.
///
/// The four outcomes must be tellable apart. Collapsing them into a boolean is what let a
/// half-downloaded planet and a damaged one look identical.
fn probe_ledger(work: &Path, report: &mut serde_json::Map<String, serde_json::Value>) {
use korp_collectors::state::{self, Completion, StateStore, Step};
let d = work.join("ledger-probe");
let _ = std::fs::create_dir_all(&d);
let mk = |name: &str, n: usize| -> PathBuf {
let p = d.join(name);
let _ = std::fs::write(&p, vec![3u8; n]);
p
};
// ── 1) An interrupted download: the file is WHOLE, nobody recorded it. ────
// The case the filesystem cannot see. `st_blocks` says "fully allocated", `st_size`
// says "full size" — and it is still not finished, because the only thing that knows
// is the producer, and the producer never got to say so.
let interrupted = mk("interrupted.osm.bz2", 1 << 20);
let st = state::judge(None, &interrupted, None, None);
report.insert("ledger.interrupted.kind".into(), st.kind().into());
report.insert("ledger.interrupted.is_fault".into(), st.is_fault().into());
report.insert(
"ledger.interrupted.looks_whole_on_disk".into(),
osm_cut::planet_looks_whole_on_disk(&interrupted).into(),
);
// ── 2) Recorded, then truncated: the two signals DISAGREE. ───────────────
let damaged = mk("damaged.osm.bz2", 1 << 20);
{
let store = match StateStore::open(&d) {
Ok(s) => s,
Err(e) => {
report.insert("ledger.fatal".into(), format!("open store: {e:#}").into());
return;
}
};
let c = Completion::of(
Step::Planet,
"damaged",
&damaged,
"src-A",
"guest-producer",
"decl-1",
)
.expect("build the completion");
if let Err(e) = store.seal(&damaged, &c) {
report.insert("ledger.fatal".into(), format!("seal: {e:#}").into());
return;
}
// Green FIRST, so the red below is a CHANGE and not the only state ever observed.
let ok = state::judge(
store.get(Step::Planet, "damaged").ok().flatten(),
&damaged,
None,
None,
);
report.insert("ledger.sealed.kind".into(), ok.kind().into());
if let Ok(f) = std::fs::OpenOptions::new().write(true).open(&damaged) {
let _ = f.set_len(1 << 19);
let _ = f.sync_all();
}
let bad = state::judge(
store.get(Step::Planet, "damaged").ok().flatten(),
&damaged,
None,
None,
);
report.insert("ledger.damaged.kind".into(), bad.kind().into());
report.insert("ledger.damaged.is_fault".into(), bad.is_fault().into());
report.insert("ledger.damaged.reason".into(), bad.reason().into());
// ── 3) A record from another source ⇒ refused, naming the difference. ─
let foreign = state::judge(
store.get(Step::Planet, "damaged").ok().flatten(),
&damaged,
Some("src-B"),
None,
);
report.insert("ledger.foreign.kind".into(), foreign.kind().into());
report.insert("ledger.foreign.reason".into(), foreign.reason().into());
}
// The writer drops here, releasing redb's exclusive flock.
// ── 4) A reader must work WHILE a writer holds the store. ────────────────
// The trap that killed five `mimir analyze` lenses. Asserted as a PAIR: a second
// writer must still be refused, and a reader must succeed anyway.
let live = d.join("live");
let _ = std::fs::create_dir_all(&live);
let art = {
let p = live.join("a.bin");
let _ = std::fs::write(&p, vec![1u8; 4096]);
p
};
match StateStore::open(&live) {
Ok(writer) => {
let c = Completion::of(Step::Planet, "planet", &art, "src", "producer", "decl")
.expect("completion");
let _ = writer.seal(&art, &c);
report.insert(
"ledger.second_writer_refused".into(),
StateStore::open(&live).is_err().into(),
);
match state::StateReader::open(&live) {
Ok(Some(r)) => {
report.insert("ledger.reader_opens_under_writer".into(), true.into());
report.insert(
"ledger.reader_sees_row".into(),
r.get(Step::Planet, "planet")
.ok()
.flatten()
.is_some()
.into(),
);
}
Ok(None) => {
report.insert("ledger.reader_opens_under_writer".into(), false.into());
}
Err(e) => {
report.insert("ledger.reader_opens_under_writer".into(), false.into());
report.insert("ledger.reader_err".into(), format!("{e:#}").into());
}
}
}
Err(e) => {
report.insert(
"ledger.fatal".into(),
format!("open live store: {e:#}").into(),
);
}
}
}
/// Step a, the torrent arm — **prove the acquisition path resolves a source without
/// pulling 176 GB.**
///
/// A cell that actually torrents a planet is a cell nobody will ever run twice, so it
/// would rot green. What is worth asserting on every run is the thing that silently
/// breaks: that the collector's declaration reaches the fetcher at all, and that a
/// planet already on disk **short-circuits the download** rather than re-fetching.
fn probe_torrent(disk_dir: &Path, report: &mut serde_json::Map<String, serde_json::Value>) {
let cfg = osm_cut::default_config();
report.insert(
"torrent.declares_planet_key".into(),
cfg.get("planet").is_some().into(),
);
// Seed a complete planet and assert acquisition is SKIPPED — the "do not download
// what you already have" leg, which is what makes the torrent path safe to leave
// wired in a test matrix.
report.insert(
"torrent.seeded_planet_short_circuits".into(),
osm_cut::cached_planet(disk_dir).is_some().into(),
);
report.insert(
"torrent.compiled_in".into(),
cfg!(feature = "osm-daemon-download").into(),
);
}
/// Which file a `damage=` token names.
fn damage_target(disk_dir: &Path, region: &str, what: &str) -> Option<PathBuf> {
let e = disk_dir.join(region);
match what {
"nodes" => Some(e.join("nodes.parquet")),
"ways" => Some(e.join("ways.parquet")),
"coords" => Some(e.join("node_coords.arrow")),
_ => None,
}
}
// ─────────────────────────────────────────────────────────────────────────────
// A disk-backed ObjectStore — so the bucket SURVIVES the power cut.
// ─────────────────────────────────────────────────────────────────────────────
// ─────────────────────────────────────────────────────────────────────────────
// Guest plumbing
// ─────────────────────────────────────────────────────────────────────────────
fn first_fixture() -> Option<PathBuf> {
let rd = std::fs::read_dir(FIXTURE_DIR).ok()?;
let mut v: Vec<PathBuf> = rd
.flatten()
.map(|e| e.path())
.filter(|p| {
let n = p.to_string_lossy().to_lowercase();
n.ends_with(".osm.bz2") || n.ends_with(".osm.pbf") || n.ends_with(".pbf")
})
.collect();
v.sort();
v.into_iter().next()
}
/// The letters of every chain step that currently holds a completion record, e.g.
/// `"abce"`. Read through the lock-free reader, exactly as korp's provisioner would.
fn recorded_steps(disk_dir: &Path, region: &str) -> String {
use korp_collectors::state::{StateReader, Step};
let Ok(Some(r)) = StateReader::open(disk_dir) else {
return String::new();
};
let mut s = String::new();
for step in Step::ALL {
let key = if step == Step::Planet {
osm_cut::PLANET_KEY
} else {
region
};
if matches!(r.get(step, key), Ok(Some(_))) {
s.push(step.letter());
}
}
s
}
fn len_of(p: &Path) -> u64 {
std::fs::metadata(p).map(|m| m.len()).unwrap_or(0)
}
/// `ARROW1` at BOTH ends — the same rule `coords_resume` applies, because a file that
/// only has the head magic is exactly what a killed writer leaves.
fn arrow_complete(p: &Path) -> bool {
let Ok(mut f) = std::fs::File::open(p) else {
return false;
};
let Ok(len) = f.metadata().map(|m| m.len()) else {
return false;
};
if len < 12 {
return false;
}
let mut head = [0u8; 6];
if f.read_exact(&mut head).is_err() || &head != b"ARROW1" {
return false;
}
if f.seek(SeekFrom::End(-6)).is_err() {
return false;
}
let mut tail = [0u8; 6];
f.read_exact(&mut tail).is_ok() && &tail == b"ARROW1"
}
fn truncate_to(p: &Path, n: u64) {
if let Ok(f) = std::fs::OpenOptions::new().write(true).open(p) {
let _ = f.set_len(n);
let _ = f.sync_all();
}
}
fn u64_json(n: u64) -> serde_json::Value {
serde_json::Value::Number(serde_json::Number::from(n))
}
fn mount(dev: &str, at: &str, fs: &str) -> Result<(), String> {
let _ = std::fs::create_dir_all(at);
let c_dev = std::ffi::CString::new(dev).map_err(|e| e.to_string())?;
let c_at = std::ffi::CString::new(at).map_err(|e| e.to_string())?;
let c_fs = std::ffi::CString::new(fs).map_err(|e| e.to_string())?;
let rc = unsafe {
libc::mount(
c_dev.as_ptr(),
c_at.as_ptr(),
c_fs.as_ptr(),
0,
std::ptr::null(),
)
};
if rc == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error().to_string())
}
}
fn mount_or_die(dev: &str, at: &str, fs: &str) {
if let Err(e) = mount(dev, at, fs) {
eprintln!("KORP-CHAIN mount {dev} -> {at} ({fs}) failed: {e}");
}
}
/// Write the report to the mailbox, echo it, and power the machine off.
fn finish(report: serde_json::Map<String, serde_json::Value>, is_pid1: bool) {
let json = serde_json::Value::Object(report).to_string();
// The mailbox write is what the host asserts on, so a failure to write it must be
// loud on the console — a silent mailbox reads exactly like a VM that never ran.
match write_mailbox(json.as_bytes()) {
Ok(()) => println!("KORP-CHAIN MAILBOX-OK {} bytes", json.len()),
Err(e) => println!("KORP-CHAIN MAILBOX-FAIL {e}"),
}
println!("KORP-CHAIN REPORT {json}");
println!("KORP-CHAIN DONE");
let _ = std::io::stdout().flush();
unsafe { libc::sync() };
if is_pid1 {
unsafe {
libc::reboot(libc::RB_POWER_OFF);
}
}
std::process::exit(0);
}
fn write_mailbox(payload: &[u8]) -> Result<(), String> {
let mut f = std::fs::OpenOptions::new()
.write(true)
.open(MAILBOX_DEV)
.map_err(|e| format!("open {MAILBOX_DEV}: {e}"))?;
let mut buf = Vec::with_capacity(8 + payload.len());
buf.extend_from_slice(&(payload.len() as u64).to_le_bytes());
buf.extend_from_slice(payload);
// Block devices want aligned, sector-sized writes.
let pad = (512 - (buf.len() % 512)) % 512;
buf.extend(std::iter::repeat_n(0u8, pad));
f.write_all(&buf).map_err(|e| format!("write: {e}"))?;
f.sync_all().map_err(|e| format!("sync: {e}"))?;
Ok(())
}