src/state.rs
//! **The collector state store — where "complete" is a RECORDED FACT, not a guess.**
//!
//! # Why this exists
//!
//! The `osm-cut` collector used to decide whether a planet was finished by looking at
//! the file: first at its NAME, then (briefly, and no better) at `st_blocks * 512 >=
//! st_size`. Both are **inferences about a state the producer already knows.** librqbit
//! knows the exact moment a torrent completes. `finalize()` knows the exact moment an
//! Arrow file is whole. Reading block allocation and hoping it correlates is guessing at
//! a proxy — and the proxy is fragile in four independent ways:
//!
//! | breaks it | how |
//! |---|---|
//! | sparse files | a hole-punched but COMPLETE file reads short |
//! | compressing filesystems (btrfs/zfs compression) | a complete file occupies FEWER blocks than its size |
//! | preallocating filesystems / delayed allocation | an INCOMPLETE file can occupy more |
//! | reflink / `cp --reflink` copies | share blocks; the accounting is not per-file |
//!
//! The `min(len)` cap the block check needed in order to survive case 3 was itself the
//! admission that the whole approach was a correlation.
//!
//! So the authority moves: **the producer writes the fact at the instant its own API
//! says done**, into an embedded ACID store, and every later question is answered from
//! that record.
//!
//! # The contract
//!
//! 1. **The producer writes it, at its own completion moment.** The torrent: librqbit's
//! completion. HTTP: content-length satisfied *and* the stream closed cleanly. The
//! convert: when `finalize()` has closed a whole file.
//! 2. **The record is written AFTER the artefact is durable** — [`StateStore::seal`]
//! fsyncs the artefact and its directory, and only then commits the row. A record
//! therefore *implies* the artefact was finished; a crash in between leaves no record,
//! which reads as "not complete". This is the discipline `coords_resume.rs` applies to
//! `node_coords.provenance` (znippy-zoomies `9ac29e7`), generalised rather than copied.
//! 3. **The record carries PROVENANCE, not a flag** — what was produced, from what
//! source, how big, its checksum if one exists, and which step of the chain it is.
//! A bare boolean would only trade a bad guess for a bad assertion.
//! 4. **The filesystem stays, as CORROBORATION, never as authority.** If the record says
//! complete and the artefact is missing or has shrunk, that is a **reported fault**
//! ([`Standing::Disagrees`]) — never a silent repair. Two independent sources that
//! disagree are information, and throwing one away is throwing the information away.
//!
//! # Concurrency — read the trap before you use this
//!
//! redb takes `flock(LOCK_EX | LOCK_NB)` on open (`redb-2.6.3`
//! `tree_store/page_store/file_backend/unix.rs:37`), so a **second opener fails
//! immediately** with `DatabaseAlreadyOpen`. Five `mimir analyze` lenses died against a
//! live server for exactly this reason.
//!
//! The collector daemon is a long-lived writer and everything else — a CLI, a matrix
//! cell, korp's provisioner — is a reader, so that is designed for here rather than
//! after the first crash: [`StateStore::open`] is the single writer and takes the
//! exclusive lock; [`StateReader::open`] opens the file **read-only and takes no lock at
//! all**, so it coexists with the running daemon. That is sound because redb commits
//! copy-on-write through an atomic commit-slot flip: a lockless reader sees either the
//! pre-commit root or the post-commit root, never a torn mixture. It cannot see a
//! half-written transaction, and it cannot block the writer.
use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use redb::{Database, ReadableTable, StorageBackend, TableDefinition};
use serde::{Deserialize, Serialize};
/// The state database, inside the collector's own `disk_dir`.
pub const STATE_DB_FILE: &str = "collector-state.redb";
/// `"<step-letter>/<key>"` → the JSON [`Completion`].
const COMPLETIONS: TableDefinition<&str, &str> = TableDefinition::new("completions");
// ─────────────────────────────────────────────────────────────────────────────
// The chain
// ─────────────────────────────────────────────────────────────────────────────
/// **A step of the a–i chain**, each of which records its own completion fact.
///
/// This mirrors `korp::pipeline::Step` deliberately and does not import it: this crate is
/// korp-free by charter (it is the leaf that breaks the facett⇄korp cycle), so the two
/// enums are peers, not a duplication one of them could delete. The LETTERS are the
/// shared contract — `korp-server --invalidate sweden:e` and a row here at step `'e'`
/// mean the same thing.
///
/// # `e` moved, and that is the whole point
///
/// It used to end at `e = Tiles`. korp's map v2 then DELETED the pyramid (`e66f846..1d53b1b`
/// removed the tiling gate, the tiled/monolith routes, `src/map_tiles.rs`, the S3 tile
/// store and 111.8 GB on disk) and korp's own chain grew to `a..g`, where `e` is
/// **networks**. Two peers that letter differently are worse than no shared contract at
/// all: `--invalidate sweden:e` and a row here at `'e'` named different artefacts, and
/// nothing could have said so. The letters are re-synchronised here, and
/// `korp::cut_driver::tests::the_leaf_and_korp_letter_the_same_chain` fails the build
/// if they ever drift again.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum Step {
/// a — a complete planet source acquired (torrent, seed, or declared path).
Planet,
/// b — convert pass 1: `nodes.parquet` + `node_coords.arrow`.
Coords,
/// c — convert pass 2: `ways.parquet` + `relations.parquet`.
Tables,
/// d — relation clip + geometry assembly: `relations.resolved.parquet`.
Relations,
/// e — road/rail/trail networks + the adjacency graph (`networks/`).
Networks,
/// f — the packed base table (native Geometry + `covering.bbox` + rung columns).
Packed,
/// g — the address index (`addr:*` → `address.parquet`).
Address,
/// h — the land polygon assembled from the region's own `natural=coastline` ways
/// (`landmass.parquet`), the map's silhouette.
Landmass,
/// i — the buildings table (`buildings/buildings.parquet`): one row per
/// `building=*` / `building:part=*` feature with a typed height, so nothing
/// downstream re-parses a height out of the `tags` blob per rebuild.
Buildings,
}
impl Step {
pub const ALL: [Step; 9] = [
Step::Planet,
Step::Coords,
Step::Tables,
Step::Relations,
Step::Networks,
Step::Packed,
Step::Address,
Step::Landmass,
Step::Buildings,
];
pub fn letter(self) -> char {
match self {
Step::Planet => 'a',
Step::Coords => 'b',
Step::Tables => 'c',
Step::Relations => 'd',
Step::Networks => 'e',
Step::Packed => 'f',
Step::Address => 'g',
Step::Landmass => 'h',
Step::Buildings => 'i',
}
}
/// The step's key — the spelling `korp::pipeline::Step::label` prints and
/// `--invalidate <region>:<key>` accepts.
pub fn label(self) -> &'static str {
match self {
Step::Planet => "planet",
Step::Coords => "coords",
Step::Tables => "tables",
Step::Relations => "relations",
Step::Networks => "networks",
Step::Packed => "packed",
Step::Address => "address",
Step::Landmass => "landmass",
Step::Buildings => "buildings",
}
}
/// What the step PRODUCES, in the operator's words — the same sentence
/// `korp::pipeline::Step::description` publishes, so one vocabulary reaches the
/// Collectors tab from either side.
pub fn description(self) -> &'static str {
match self {
Step::Planet => "planet source on disk",
Step::Coords => "convert pass 1 (nodes.parquet + node coord store)",
Step::Tables => "convert pass 2 (ways/relations)",
Step::Relations => "relation clip + geometry assembly",
Step::Networks => "road/rail/trail networks + adjacency graph",
Step::Packed => "packed base table (covering.bbox + rung columns)",
Step::Address => "address index (addr:* → address.parquet)",
Step::Landmass => "land polygon from the region's own coastline ways (→ landmass.parquet)",
Step::Buildings => "buildings table (height_m / levels / kind → buildings/buildings.parquet)",
}
}
pub fn parse(s: &str) -> Option<Step> {
match s.trim().to_ascii_lowercase().as_str() {
"a" | "planet" => Some(Step::Planet),
"b" | "coords" | "pass1" | "nodes" => Some(Step::Coords),
"c" | "tables" | "pass2" | "ways" => Some(Step::Tables),
"d" | "relations" | "resolved" => Some(Step::Relations),
"e" | "networks" | "graph" => Some(Step::Networks),
"f" | "packed" | "pack" | "base" => Some(Step::Packed),
"g" | "address" | "addr" | "geocode" => Some(Step::Address),
"h" | "landmass" | "coastline" | "land" => Some(Step::Landmass),
"i" | "buildings" | "building" | "height" => Some(Step::Buildings),
_ => None,
}
}
}
/// The row key: `"<letter>/<key>"`, e.g. `"a/planet"` or `"e/sweden"`.
fn row_key(step: Step, key: &str) -> String {
format!("{}/{key}", step.letter())
}
// ─────────────────────────────────────────────────────────────────────────────
// The fact
// ─────────────────────────────────────────────────────────────────────────────
/// **What one producer finished, and everything needed to judge it later.**
///
/// Deliberately not a boolean. "Complete" on its own cannot answer *"complete from
/// WHICH planet"* or *"complete under WHICH declaration"*, and those are the two
/// questions that decide whether the next run may reuse the artefact or must reproduce
/// it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Completion {
/// Which step of a–e finished.
pub step: Step,
/// What finished (a region name, or `"planet"`).
pub key: String,
/// The artefact on disk, as an absolute path when one is knowable.
pub artefact: String,
/// **What it was produced FROM** — the planet for a convert, the torrent/seed for a
/// planet, the extract for a pyramid. A record whose source does not match what the
/// caller now has is [`Standing::Foreign`] and is refused BY NAME.
pub source: String,
/// The artefact's size at the moment the producer declared it finished. Compared
/// against the file later: shrinkage is a fault, not a reason to re-cut quietly.
pub bytes: u64,
/// A checksum, when the producer had one to give. `None` is honest — the OSM planet
/// distribution publishes no sidecar on this box — and never a fabricated digest.
pub checksum: Option<String>,
/// **Who declared it done**, e.g. `"librqbit (osm_katana::fixture::ensure_planet_osm)"`.
/// This is the field that makes the record auditable: it names the API whose word we
/// took.
pub producer: String,
/// The declaration in force when it was produced (zooms, clip, revision). A changed
/// declaration makes an otherwise-perfect artefact [`Standing::Foreign`].
pub declaration: String,
/// When the producer said done (unix seconds).
pub completed_unix: u64,
}
impl Completion {
/// A record for `step`/`key`, sized from the artefact ITSELF at the moment of
/// sealing — never from a number the caller carried in, which could describe a file
/// that was never written.
pub fn of(
step: Step,
key: impl Into<String>,
artefact: &Path,
source: impl Into<String>,
producer: impl Into<String>,
declaration: impl Into<String>,
) -> Result<Self> {
let md = fs::metadata(artefact)
.with_context(|| format!("stat the artefact being sealed: {}", artefact.display()))?;
Ok(Self {
step,
key: key.into(),
artefact: artefact.display().to_string(),
source: source.into(),
bytes: md.len(),
checksum: None,
producer: producer.into(),
declaration: declaration.into(),
completed_unix: now_unix(),
})
}
pub fn with_checksum(mut self, sum: impl Into<String>) -> Self {
self.checksum = Some(sum.into());
self
}
/// A one-line human rendering for a log or a failure message.
pub fn line(&self) -> String {
format!(
"{}({}) {} — {} bytes from `{}` by `{}` under `{}`{}",
self.step.letter(),
self.key,
self.artefact,
self.bytes,
self.source,
self.producer,
self.declaration,
self.checksum
.as_ref()
.map(|c| format!(" [{c}]"))
.unwrap_or_default(),
)
}
}
fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
// ─────────────────────────────────────────────────────────────────────────────
// The verdict
// ─────────────────────────────────────────────────────────────────────────────
/// **What the record and the artefact TOGETHER say.**
///
/// Only [`Standing::Complete`] is green. The other three are all different, and
/// collapsing them into `false` is what made the old check unable to tell "still
/// downloading" from "the disk lost my data".
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Standing {
/// **No record.** The producer never said it finished — it is still running, or it
/// died before it got there. Not an error: reproduce it.
Unrecorded { why: String },
/// **Recorded, and the artefact corroborates.** The only green.
Complete(Box<Completion>),
/// **Recorded, but the artefact disagrees** — gone, or smaller than the producer
/// sealed. Something outside this collector damaged it. A FAULT to report, never a
/// thing to quietly re-cut: silently repairing it hides that the disk, or another
/// process, is eating artefacts.
Disagrees {
record: Box<Completion>,
why: String,
},
/// **Recorded, but from a different source or declaration.** Refused, naming the
/// difference, so an operator can see WHICH of the two changed.
Foreign {
record: Box<Completion>,
why: String,
},
}
impl Standing {
/// Green — and nothing else is.
pub fn is_complete(&self) -> bool {
matches!(self, Standing::Complete(_))
}
/// A fault the operator must see (as opposed to work still to do).
pub fn is_fault(&self) -> bool {
matches!(self, Standing::Disagrees { .. })
}
pub fn reason(&self) -> String {
match self {
Standing::Unrecorded { why } => why.clone(),
Standing::Complete(c) => format!("recorded complete: {}", c.line()),
Standing::Disagrees { why, .. } | Standing::Foreign { why, .. } => why.clone(),
}
}
/// The stable token a test or a matrix cell asserts on.
pub fn kind(&self) -> &'static str {
match self {
Standing::Unrecorded { .. } => "unrecorded",
Standing::Complete(_) => "complete",
Standing::Disagrees { .. } => "disagrees",
Standing::Foreign { .. } => "foreign",
}
}
}
/// **Judge a record against the artefact it claims** — the ONE place the two sources are
/// combined, so no caller can accidentally trust only one of them.
///
/// `want_source`/`want_declaration` are `None` when the caller has nothing to compare
/// against (a bare completeness question); `Some` when it does, and then a mismatch is
/// [`Standing::Foreign`].
///
/// The size test is `on_disk < recorded` — **shrinkage only**. A file that GREW since it
/// was sealed is an append by something else, which is a different fault and not one this
/// function can attribute; it is reported through the same arm with its own wording
/// rather than being silently accepted.
pub fn judge(
record: Option<Completion>,
artefact: &Path,
want_source: Option<&str>,
want_declaration: Option<&str>,
) -> Standing {
let Some(rec) = record else {
return Standing::Unrecorded {
why: format!(
"no completion record for {} — the producer never declared it finished \
(still running, or it died before sealing)",
artefact.display()
),
};
};
if let Some(want) = want_source {
if rec.source != want {
let why = format!(
"the record was produced from `{}`, the caller now has `{}` ⇒ REFUSING to \
reuse it",
rec.source, want
);
return Standing::Foreign {
record: Box::new(rec),
why,
};
}
}
if let Some(want) = want_declaration {
if rec.declaration != want {
let why = format!(
"the record was produced under declaration `{}`, now `{}` ⇒ REFUSING to \
reuse it",
rec.declaration, want
);
return Standing::Foreign {
record: Box::new(rec),
why,
};
}
}
match fs::metadata(artefact) {
Err(e) => {
let why = format!(
"the record says {} was completed at {} bytes by `{}`, but it cannot be \
read now: {e}. The record and the disk DISAGREE — reporting rather than \
quietly re-cutting, because something outside this collector removed a \
sealed artefact",
artefact.display(),
rec.bytes,
rec.producer
);
Standing::Disagrees {
record: Box::new(rec),
why,
}
}
Ok(md) if md.len() < rec.bytes => {
let why = format!(
"the record says {} was completed at {} bytes by `{}`; it is {} bytes now \
({} short). The record and the disk DISAGREE — a sealed artefact has \
SHRUNK, which no step of this chain does",
artefact.display(),
rec.bytes,
rec.producer,
md.len(),
rec.bytes - md.len()
);
Standing::Disagrees {
record: Box::new(rec),
why,
}
}
Ok(md) if md.len() > rec.bytes => {
let why = format!(
"the record says {} was completed at {} bytes by `{}`; it is {} bytes now \
({} MORE). Something appended to a sealed artefact — reported, not \
accepted",
artefact.display(),
rec.bytes,
rec.producer,
md.len(),
md.len() - rec.bytes
);
Standing::Disagrees {
record: Box::new(rec),
why,
}
}
Ok(_) => Standing::Complete(Box::new(rec)),
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The writer
// ─────────────────────────────────────────────────────────────────────────────
/// **The single writer.** Holds redb's exclusive lock for as long as it lives, so the
/// collector daemon owns one and nothing else opens one.
#[derive(Debug)]
pub struct StateStore {
db: Database,
path: PathBuf,
}
impl StateStore {
/// Open (creating if absent) the store under `disk_dir`.
///
/// Fails loudly if another process holds it. That is the correct answer — two writers
/// to a completion ledger is exactly the situation nobody wants resolved silently —
/// and the message says so rather than surfacing redb's bare `DatabaseAlreadyOpen`.
pub fn open(disk_dir: &Path) -> Result<Self> {
fs::create_dir_all(disk_dir).with_context(|| format!("mkdir {}", disk_dir.display()))?;
let path = disk_dir.join(STATE_DB_FILE);
let db = Database::create(&path).with_context(|| {
format!(
"open the collector state store at {} for WRITING. redb takes an exclusive \
flock, so this fails while another writer holds it — there must be exactly \
one. Readers must use StateReader::open, which takes no lock",
path.display()
)
})?;
// Materialise the table so a reader opening a brand-new store finds it rather
// than erroring on a missing table.
let w = db.begin_write()?;
{
let _ = w.open_table(COMPLETIONS)?;
}
w.commit()?;
Ok(Self { db, path })
}
pub fn path(&self) -> &Path {
&self.path
}
/// **Seal a completion: make the artefact durable, THEN record it.**
///
/// The order is the whole point. `fsync` the artefact (and its directory, so the
/// name is durable too, not just the bytes), and only then commit the row. A crash
/// anywhere before the commit leaves NO record, which reads as "not complete" and is
/// reproduced — never a record pointing at a file that never landed.
///
/// Reversing these two lines would reintroduce precisely the bug this module exists
/// to remove, in a worse form: a durable claim about a non-durable file.
pub fn seal(&self, artefact: &Path, c: &Completion) -> Result<()> {
fsync_file_and_parent(artefact)
.with_context(|| format!("fsync {} before sealing it", artefact.display()))?;
self.record(c)
}
/// Write the row with no fsync of the artefact.
///
/// For producers whose artefact is not a single file (a tile pyramid is thousands of
/// objects, some of them in an object store that fsync means nothing to). Those
/// callers must have made their output durable by their own means FIRST; the
/// ordering contract is theirs to keep, and naming it in a separate method is what
/// makes that visible at the call site instead of hiding it behind `seal`.
pub fn record(&self, c: &Completion) -> Result<()> {
let json = serde_json::to_string(c).context("serialize the completion record")?;
let key = row_key(c.step, &c.key);
let w = self.db.begin_write()?;
{
let mut t = w.open_table(COMPLETIONS)?;
t.insert(key.as_str(), json.as_str())?;
}
w.commit().context("commit the completion record")?;
Ok(())
}
/// **Forget a completion — the "nuke it" half of the chain's contract.**
///
/// Removing the RECORD is what makes a step incomplete; it is not a side effect of
/// deleting files. A caller that deletes an artefact and leaves its record standing
/// creates a [`Standing::Disagrees`] fault on the next pass, which is correct
/// behaviour and exactly why this must be called deliberately.
pub fn forget(&self, step: Step, key: &str) -> Result<bool> {
let k = row_key(step, key);
let w = self.db.begin_write()?;
let existed;
{
let mut t = w.open_table(COMPLETIONS)?;
existed = t.remove(k.as_str())?.is_some();
}
w.commit()?;
Ok(existed)
}
/// Forget `from` and every step after it — the cascade `--invalidate <region>:<step>`
/// means. Returns the steps that actually had a record.
pub fn invalidate_from(&self, from: Step, key: &str) -> Result<Vec<Step>> {
let mut gone = Vec::new();
for s in Step::ALL.into_iter().filter(|s| *s >= from) {
if self.forget(s, key)? {
gone.push(s);
}
}
Ok(gone)
}
pub fn get(&self, step: Step, key: &str) -> Result<Option<Completion>> {
let r = self.db.begin_read()?;
let t = r.open_table(COMPLETIONS)?;
read_one(&t, step, key)
}
/// Every record, ordered by step then key — for a status verb or a matrix cell.
pub fn all(&self) -> Result<Vec<Completion>> {
let r = self.db.begin_read()?;
let t = r.open_table(COMPLETIONS)?;
let mut out = Vec::new();
for row in t.iter()? {
let (_, v) = row?;
if let Ok(c) = serde_json::from_str::<Completion>(v.value()) {
out.push(c);
}
}
out.sort_by(|a, b| (a.step, &a.key).cmp(&(b.step, &b.key)));
Ok(out)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The reader — no lock, so it coexists with the running daemon
// ─────────────────────────────────────────────────────────────────────────────
/// How many times a snapshot read is retried when the writer commits underneath it.
const SNAPSHOT_ATTEMPTS: usize = 8;
/// The prefix that carries redb's commit slots — the bytes whose change means "a
/// transaction landed while we were reading".
const HEADER_BYTES: usize = 4096;
/// A redb backend serving a **RAM snapshot of the file, which it never writes back**.
///
/// # Why not simply open the file read-only
///
/// Two dead ends, both measured against `redb-2.6.3`:
///
/// * redb's stock `FileBackend` takes `flock(LOCK_EX | LOCK_NB)`
/// (`tree_store/page_store/file_backend/unix.rs:37`), so a second opener gets
/// `DatabaseAlreadyOpen` — the failure that killed five `mimir analyze` lenses against
/// a live server. `Builder::open` uses that same backend, so it is no escape. A
/// *shared* lock is no escape either: `LOCK_SH` against the daemon's held `LOCK_EX`
/// still returns `WouldBlock`, so a reader could only ever run when the thing it wants
/// to read is stopped — which is never when anybody asks.
/// * `create_with_backend` is the only constructor that accepts a custom backend, and it
/// passes redb's `allow_create` flag, so **redb writes to the header during open**. A
/// backend that refuses writes therefore cannot be opened at all: the first version of
/// this type returned `PermissionDenied` and failed its own test.
///
/// So the snapshot is taken into memory and redb is allowed to scribble on *that*. The
/// file on disk is opened `O_RDONLY`, is never locked, and is never written.
///
/// # The guarantee, and its limit — stated because it is not absolute
///
/// The copy is not atomic, so it is taken under a **seqlock**: read the header, read the
/// body, read the header again; if it moved, a transaction committed underneath us and
/// the read is retried (up to [`SNAPSHOT_ATTEMPTS`]). A snapshot that survives that check
/// saw no commit between its first and last byte.
///
/// The limit: this detects a *concurrent commit*, not every possible page recycle, and a
/// store being rewritten continuously could in principle exhaust the retries — in which
/// case this returns an **error naming that**, rather than handing back a torn database.
/// For a completion ledger — a handful of rows, written once per finished step — the
/// retry path is close to unreachable, and it is a bounded failure rather than silent
/// corruption if it ever is not.
///
/// # One writer of this behaviour
///
/// Every lock-free reader of a redb store in this crate opens through
/// [`open_snapshot_db`], which is the only constructor of this backend. The completion
/// ledger ([`StateReader`]) and the live register
/// ([`crate::register::RegisterReader`](crate::register)) are two *questions*, not two
/// mechanisms — a second copy of the seqlock would be a second set of retry bounds to
/// keep in agreement.
#[derive(Debug)]
pub(crate) struct SnapshotBackend {
/// The whole file as of the snapshot, plus whatever redb writes during open. Never
/// flushed anywhere.
data: std::sync::Mutex<Vec<u8>>,
}
impl SnapshotBackend {
fn open(path: &Path) -> io::Result<Self> {
for attempt in 1..=SNAPSHOT_ATTEMPTS {
let before = read_prefix(path, HEADER_BYTES)?;
let body = fs::read(path)?;
let after = read_prefix(path, HEADER_BYTES)?;
if before == after {
return Ok(Self {
data: std::sync::Mutex::new(body),
});
}
// Back off a touch so a burst of commits is not met with a tight spin.
std::thread::sleep(std::time::Duration::from_millis(2 * attempt as u64));
}
Err(io::Error::new(
io::ErrorKind::WouldBlock,
format!(
"could not take a stable snapshot of {} in {SNAPSHOT_ATTEMPTS} attempts — the \
writer committed during every one. Refusing to open a possibly-torn \
database rather than serving one",
path.display()
),
))
}
}
fn read_prefix(path: &Path, n: usize) -> io::Result<Vec<u8>> {
use std::io::Read;
let mut f = fs::File::open(path)?;
let mut buf = vec![0u8; n];
let mut got = 0;
while got < n {
match f.read(&mut buf[got..])? {
0 => break,
k => got += k,
}
}
buf.truncate(got);
Ok(buf)
}
impl StorageBackend for SnapshotBackend {
fn len(&self) -> io::Result<u64> {
Ok(self.data.lock().map_err(poisoned)?.len() as u64)
}
fn read(&self, offset: u64, len: usize) -> io::Result<Vec<u8>> {
let d = self.data.lock().map_err(poisoned)?;
let start = offset as usize;
let end = start.saturating_add(len);
if end > d.len() {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
format!("read {start}..{end} past the {}-byte snapshot", d.len()),
));
}
Ok(d[start..end].to_vec())
}
fn set_len(&self, len: u64) -> io::Result<()> {
self.data.lock().map_err(poisoned)?.resize(len as usize, 0);
Ok(())
}
fn sync_data(&self, _eventual: bool) -> io::Result<()> {
// Nothing is going to disk, so there is nothing to flush — and this must SUCCEED,
// because redb calls it on paths a reader legitimately reaches.
Ok(())
}
fn write(&self, offset: u64, data: &[u8]) -> io::Result<()> {
let mut d = self.data.lock().map_err(poisoned)?;
let end = offset as usize + data.len();
if end > d.len() {
d.resize(end, 0);
}
d[offset as usize..end].copy_from_slice(data);
Ok(())
}
}
fn poisoned<T>(_: std::sync::PoisonError<T>) -> io::Error {
io::Error::other("the state-store snapshot mutex was poisoned")
}
/// **Open any redb file in this crate for LOCK-FREE reading**, as a RAM snapshot that is
/// never written back.
///
/// `Ok(None)` when the file does not exist — a box where the writer has never run is an
/// answer, not a failure, and every caller wants to say so in its own words.
///
/// This is the seam that lets a *separate process* read a store the running service
/// holds open for writing. It is one function rather than one per store, because the
/// hazard it navigates (redb's `flock(LOCK_EX)`, and `create_with_backend` writing to the
/// header during open) is a property of redb, not of any particular table.
pub(crate) fn open_snapshot_db(path: &Path) -> Result<Option<Database>> {
if !path.is_file() {
return Ok(None);
}
let backend = SnapshotBackend::open(path)
.with_context(|| format!("snapshot {} for reading", path.display()))?;
let db = Database::builder()
.create_with_backend(backend)
.with_context(|| format!("open {} for reading (lock-free snapshot)", path.display()))?;
Ok(Some(db))
}
/// **A lock-free reader.** Safe to open while the daemon is writing.
#[derive(Debug)]
pub struct StateReader {
db: Database,
}
impl StateReader {
/// Open the store for reading. `Ok(None)` when there is no store yet — a box where
/// the collector has never run is not an error, and must not read as one.
pub fn open(disk_dir: &Path) -> Result<Option<Self>> {
let path = disk_dir.join(STATE_DB_FILE);
Ok(open_snapshot_db(&path)?.map(|db| Self { db }))
}
pub fn get(&self, step: Step, key: &str) -> Result<Option<Completion>> {
let r = self.db.begin_read()?;
let t = match r.open_table(COMPLETIONS) {
Ok(t) => t,
// A store that exists but has no table yet holds no completions. That is an
// answer, not a failure.
Err(redb::TableError::TableDoesNotExist(_)) => return Ok(None),
Err(e) => return Err(e.into()),
};
read_one(&t, step, key)
}
/// [`judge`] against this reader's record, in one call — the shape every consumer
/// wants, so none of them has to remember to combine the two sources itself.
pub fn standing(
&self,
step: Step,
key: &str,
artefact: &Path,
want_source: Option<&str>,
want_declaration: Option<&str>,
) -> Result<Standing> {
Ok(judge(
self.get(step, key)?,
artefact,
want_source,
want_declaration,
))
}
}
fn read_one(
t: &impl ReadableTable<&'static str, &'static str>,
step: Step,
key: &str,
) -> Result<Option<Completion>> {
let k = row_key(step, key);
match t.get(k.as_str())? {
None => Ok(None),
Some(v) => Ok(serde_json::from_str::<Completion>(v.value()).ok()),
}
}
/// fsync the file AND the directory holding it.
///
/// Syncing only the file leaves the directory entry unsynced, so a crash can lose the
/// NAME of a file whose bytes are safely on the platter — the artefact would be durable
/// and unreachable, and the record would point at nothing.
fn fsync_file_and_parent(p: &Path) -> io::Result<()> {
fs::File::open(p)?.sync_all()?;
if let Some(dir) = p.parent() {
// A directory fsync needs the directory opened; failure to open it (it is a
// relative path with no parent, say) is not fatal to the file's own durability.
if let Ok(d) = fs::File::open(dir) {
let _ = d.sync_all();
}
}
Ok(())
}
// ─────────────────────────────────────────────────────────────────────────────
// RED PROBES — the three ways this contract must be able to fail
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
fn art(dir: &Path, name: &str, bytes: usize) -> PathBuf {
let p = dir.join(name);
fs::write(&p, vec![7u8; bytes]).unwrap();
p
}
fn seal_one(dir: &Path, p: &Path, source: &str, decl: &str) -> StateStore {
let s = StateStore::open(dir).unwrap();
let c = Completion::of(Step::Planet, "planet", p, source, "test-producer", decl).unwrap();
s.seal(p, &c).unwrap();
s
}
/// The green, so the reds below mean something: a sealed artefact reads COMPLETE.
#[test]
fn a_sealed_artefact_reads_complete_and_carries_its_provenance() {
let d = tempfile::tempdir().unwrap();
let p = art(d.path(), "planet.osm.bz2", 4096);
let s = seal_one(d.path(), &p, "torrent:magnet:xyz", "rev=1");
let st = judge(
s.get(Step::Planet, "planet").unwrap(),
&p,
Some("torrent:magnet:xyz"),
Some("rev=1"),
);
assert!(st.is_complete(), "{}", st.reason());
let Standing::Complete(c) = st else {
unreachable!()
};
assert_eq!(c.bytes, 4096);
assert_eq!(c.source, "torrent:magnet:xyz");
assert_eq!(c.producer, "test-producer");
assert!(
c.checksum.is_none(),
"a checksum nobody computed must stay None, never a fabricated digest"
);
}
/// ★ **RED 1 — an interrupted download leaves NO record, so the next run redoes it.**
///
/// The artefact is present and, crucially, would satisfy every filesystem heuristic:
/// it is a fully-written, fully-allocated file. Only the absence of the producer's
/// word makes it incomplete — which is the entire point of moving the authority.
#[test]
fn an_interrupted_download_has_no_record_and_is_therefore_incomplete() {
let d = tempfile::tempdir().unwrap();
let p = art(d.path(), "planet.osm.bz2", 8192);
let s = StateStore::open(d.path()).unwrap();
// Producer died before sealing.
let st = judge(s.get(Step::Planet, "planet").unwrap(), &p, None, None);
assert_eq!(st.kind(), "unrecorded", "{}", st.reason());
assert!(!st.is_complete());
assert!(
!st.is_fault(),
"an unfinished download is work to do, NOT a fault to report"
);
assert!(
st.reason().contains("never declared it finished"),
"the reason must name WHY: {}",
st.reason()
);
}
/// ★ **RED 2 — record present, artefact truncated ⇒ a REPORTED FAULT, not silent green
/// and not a silent re-cut.**
#[test]
fn a_record_over_a_truncated_artefact_disagrees_and_is_reported_as_a_fault() {
let d = tempfile::tempdir().unwrap();
let p = art(d.path(), "planet.osm.bz2", 8192);
let s = seal_one(d.path(), &p, "seed", "rev=1");
// Something outside the collector eats half the file.
fs::OpenOptions::new()
.write(true)
.open(&p)
.unwrap()
.set_len(4096)
.unwrap();
let st = judge(s.get(Step::Planet, "planet").unwrap(), &p, None, None);
assert_eq!(st.kind(), "disagrees", "{}", st.reason());
assert!(!st.is_complete());
assert!(
st.is_fault(),
"a shrunken sealed artefact MUST surface as a fault"
);
assert!(st.reason().contains("SHRUNK"), "{}", st.reason());
assert!(
st.reason().contains("4096"),
"the reason must carry both numbers: {}",
st.reason()
);
// And the same for an artefact that vanished entirely.
fs::remove_file(&p).unwrap();
let gone = judge(s.get(Step::Planet, "planet").unwrap(), &p, None, None);
assert_eq!(gone.kind(), "disagrees", "{}", gone.reason());
assert!(gone.is_fault());
}
/// ★ **RED 3 — a record from another source, or another declaration, is REFUSED BY
/// NAME.**
#[test]
fn a_record_from_another_extract_is_refused_and_the_reason_names_the_difference() {
let d = tempfile::tempdir().unwrap();
let p = art(d.path(), "planet.osm.bz2", 4096);
let s = seal_one(d.path(), &p, "torrent:planet-260427", "zooms=6,9;rev=1");
let other_source = judge(
s.get(Step::Planet, "planet").unwrap(),
&p,
Some("torrent:planet-260501"),
None,
);
assert_eq!(other_source.kind(), "foreign", "{}", other_source.reason());
assert!(
other_source.reason().contains("planet-260427"),
"{}",
other_source.reason()
);
assert!(
other_source.reason().contains("planet-260501"),
"{}",
other_source.reason()
);
let other_decl = judge(
s.get(Step::Planet, "planet").unwrap(),
&p,
None,
Some("zooms=6,9,12;rev=1"),
);
assert_eq!(other_decl.kind(), "foreign", "{}", other_decl.reason());
assert!(
other_decl.reason().contains("zooms=6,9,12"),
"{}",
other_decl.reason()
);
// …and the matching pair is still green, or the two above prove nothing.
let same = judge(
s.get(Step::Planet, "planet").unwrap(),
&p,
Some("torrent:planet-260427"),
Some("zooms=6,9;rev=1"),
);
assert!(same.is_complete(), "{}", same.reason());
}
/// ★ **The reader must work WHILE the writer holds the store** — the trap that killed
/// five `mimir analyze` lenses.
///
/// Both halves are asserted, because only the pair is evidence: a second WRITER must
/// still be refused (one writer is the invariant), and a READER must succeed anyway.
#[test]
fn a_reader_opens_while_the_writer_holds_the_exclusive_lock() {
let d = tempfile::tempdir().unwrap();
let p = art(d.path(), "planet.osm.bz2", 2048);
let writer = seal_one(d.path(), &p, "seed", "rev=1");
assert!(
StateStore::open(d.path()).is_err(),
"a SECOND writer was allowed — one writer is the whole invariant"
);
let reader = StateReader::open(d.path())
.expect("open the reader")
.expect("the store exists");
let got = reader
.get(Step::Planet, "planet")
.unwrap()
.expect("the reader sees the row");
assert_eq!(got.bytes, 2048);
// A commit made AFTER the reader opened must be visible to a fresh reader — the
// lockless snapshot has to advance, or readers would silently serve stale facts.
let p2 = art(d.path(), "ways.parquet", 512);
let c2 = Completion::of(
Step::Tables,
"fixture",
&p2,
"planet.osm.bz2",
"convert",
"rev=1",
)
.unwrap();
writer.seal(&p2, &c2).unwrap();
let reader2 = StateReader::open(d.path()).unwrap().unwrap();
assert!(
reader2.get(Step::Tables, "fixture").unwrap().is_some(),
"a reader opened after the commit cannot see it — the snapshot never advanced"
);
}
/// Forgetting a record is what makes a step incomplete, and the cascade drops every
/// later step with it.
#[test]
fn invalidating_a_step_forgets_it_and_everything_after_it() {
let d = tempfile::tempdir().unwrap();
let s = StateStore::open(d.path()).unwrap();
for step in Step::ALL {
let p = art(d.path(), &format!("{}.bin", step.letter()), 128);
let c = Completion::of(step, "sweden", &p, "src", "prod", "decl").unwrap();
s.seal(&p, &c).unwrap();
}
assert_eq!(s.all().unwrap().len(), Step::ALL.len());
let gone = s.invalidate_from(Step::Tables, "sweden").unwrap();
// The TAIL is derived from the chain's own order, not spelled out. It was
// spelled out, and it went stale the day step h (`Landmass`) joined `Step::ALL`
// in f7aa5c8 — the cascade was right, the expectation was wrong, and this test
// has been red on main ever since. A literal list of steps in a test about the
// step list is a copy that only ever gets one of them updated.
//
// What the test is actually FOR is the cut point, and that is still asserted
// literally below: `b` survives, `c` and everything after it does not.
let expected: Vec<Step> = Step::ALL
.iter()
.copied()
.skip_while(|s| *s != Step::Tables)
.collect();
assert_eq!(gone, expected);
assert_eq!(
gone.len(),
Step::ALL.len() - 2,
"everything but the two steps BEFORE the cut (planet, coords) must go",
);
assert!(
s.get(Step::Coords, "sweden").unwrap().is_some(),
"b must survive an invalidate from c"
);
assert!(s.get(Step::Tables, "sweden").unwrap().is_none());
assert!(s.get(Step::Address, "sweden").unwrap().is_none());
// Forgetting an absent record reports false rather than pretending it removed one.
assert!(!s.forget(Step::Address, "sweden").unwrap());
}
/// ★ The reader's writes must land in RAM and **never reach the file**.
///
/// redb writes to the header while opening, so the backend cannot refuse writes
/// outright (the first version did, and failed to open at all). What it must
/// guarantee instead is that those writes go nowhere: the bytes on disk are
/// identical before and after a reader has opened, read and dropped.
#[test]
fn the_reader_never_writes_a_byte_back_to_the_file() {
let d = tempfile::tempdir().unwrap();
let p = art(d.path(), "x.bin", 16);
let w = seal_one(d.path(), &p, "s", "d");
let db_path = d.path().join(STATE_DB_FILE);
let before = fs::read(&db_path).unwrap();
{
let r = StateReader::open(d.path()).unwrap().unwrap();
assert!(r.get(Step::Planet, "planet").unwrap().is_some());
}
let after = fs::read(&db_path).unwrap();
assert_eq!(
before, after,
"the reader modified the store on disk — it is supposed to scribble only on \
its own RAM snapshot"
);
// And the writer is still perfectly usable afterwards.
let p2 = art(d.path(), "y.bin", 32);
let c = Completion::of(Step::Address, "r", &p2, "s", "prod", "d").unwrap();
w.seal(&p2, &c).unwrap();
assert!(w.get(Step::Address, "r").unwrap().is_some());
}
}