//! **`Acquire` — the seam for artefacts too big to hold in memory** (feature `daemon`).
//!
//! # Why this is a SECOND seam and not a wider `Fetcher`
//!
//! [`Fetcher`](crate::daemon::Fetcher) is `Fn(&str) -> Vec<u8>`: the whole body in RAM.
//! For a 30 MB DCAT dump that is exactly right, and twenty of the collectors in this
//! crate want nothing more. Widening it — a destination path, resumption, partial state,
//! progress — would make those twenty pay for one collector's problem, at every call
//! site.
//!
//! So there are two seams and **the discriminator is size, not transport**. Something
//! that fits in memory uses `Fetcher`. Something that does not comes through here, and
//! gets what a large fetch actually needs:
//!
//! | a big fetch needs | `Fetcher` | `Acquire` |
//! |---|---|---|
//! | never holds the body in RAM | — | streams to `dest` |
//! | resumes an interrupted transfer | — | `.partial` + `Range:` |
//! | reports progress somewhere the supervisor can see | — | [`AcquireSink::progress`] |
//! | can be told to stop | — | [`Shutdown`](crate::daemon::Shutdown) |
//! | says whether the result is COMPLETE, per format | — | [`Completeness`] |
//! | is idempotent over an already-landed artefact | — | no network on the second call |
//!
//! # What it replaces
//!
//! One download written three times, which is the defect this seam exists to end:
//!
//! ```text
//! osm_katana::fixture korp/src/download.rs region.rs::download_to
//! librqbit body, own tokio du-vs-ls probe, exists resp.bytes() → RAM,
//! runtime, progress to its only because the first then .partial + rename
//! OWN stderr bar cannot be observed
//! ```
//!
//! [`HttpRange`] is the third of those, replaced rather than wrapped: `region.rs` now
//! calls it, and the body never enters RAM. The librqbit body is the next move (design
//! §5 phase 3) and is deliberately not here yet — a torrent implementation belongs
//! behind an `acquire-torrent` feature with the constants staying in osm-katana, and
//! declaring that feature before writing it would be a name with nothing behind it.
//!
//! # Completeness is per format, and that is on purpose
//!
//! [`Completeness`] is an enum, not a `bool`. Arrow and Parquet have footers, so a
//! truncated file cannot lie about being whole; a JSON body, a CSV or a torrent has no
//! such marker. A single generic "is it complete" predicate would be a check that cannot
//! fail — so the caller declares which check applies, and a caller that has no marker
//! says [`Completeness::Any`] out loud rather than getting a false green for free.
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result, bail};
use crate::daemon::{LogSink, Shutdown};
/// Bytes moved between progress callbacks and cancel checks. Big enough that a 176 GB
/// transfer is not one callback per packet, small enough that a stop is observed in well
/// under a second on any link that is moving at all.
const CHUNK: usize = 1 << 20;
// ─────────────────────────────────────────────────────────────────────────────
// The request
// ─────────────────────────────────────────────────────────────────────────────
/// Where the bytes come from. One variant per **transport**, because that is what differs
/// operationally; size is what decides `Fetcher`-vs-`Acquire` and is already decided by
/// the time a caller is here.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Source {
/// An HTTP(S) URL, fetched resumably.
Http { url: String },
/// **An explicitly DECLARED local path.** Not a search, not a home directory: the
/// operator names the file. See [`LocalSeed`] for why that distinction is the whole
/// point of the variant.
LocalSeed { declared: Vec<PathBuf> },
}
/// **How the caller knows the artefact is whole.** Per format, never generic — see the
/// module note.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Completeness {
/// No marker exists for this payload. Said out loud, so a reader of the call site
/// knows the transfer's own framing (content-length, a clean stream close) is all the
/// assurance there is.
Any,
/// At least this many bytes. The honest check for a source that publishes no digest
/// and no exact length — it catches a truncation, and it does not pretend to catch a
/// corruption.
MinBytes(u64),
/// Exactly this many bytes.
ExactBytes(u64),
/// A blake3 hex digest. The fleet's hash (znippy, skade), so this is the same crate
/// at the same major rather than a second hashing opinion.
Blake3(String),
}
impl Completeness {
/// Does `path` satisfy this declaration? `Ok(())` or a message naming what fell short.
///
/// The digest arm reads the file in [`CHUNK`] blocks — a 176 GB artefact must be
/// verifiable without being loaded, or the verification reintroduces exactly the
/// memory ceiling this module exists to remove.
pub fn check(&self, path: &Path) -> Result<()> {
let md = std::fs::metadata(path)
.with_context(|| format!("stat {} to check completeness", path.display()))?;
match self {
Completeness::Any => Ok(()),
Completeness::MinBytes(min) => {
if md.len() < *min {
bail!(
"{} is {} bytes, short of the declared minimum {min} — INCOMPLETE",
path.display(),
md.len()
);
}
Ok(())
}
Completeness::ExactBytes(want) => {
if md.len() != *want {
bail!(
"{} is {} bytes, not the declared {want} — INCOMPLETE",
path.display(),
md.len()
);
}
Ok(())
}
Completeness::Blake3(want) => {
let got = blake3_file(path)?;
if !got.eq_ignore_ascii_case(want) {
bail!(
"{} hashes blake3:{got}, the declaration says blake3:{want} — the bytes \
are not the bytes that were asked for",
path.display()
);
}
Ok(())
}
}
}
/// A short rendering for a log line or a provenance record.
pub fn label(&self) -> String {
match self {
Completeness::Any => "no completeness marker declared".into(),
Completeness::MinBytes(n) => format!("≥{n} bytes"),
Completeness::ExactBytes(n) => format!("={n} bytes"),
Completeness::Blake3(h) => format!("blake3:{h}"),
}
}
}
/// Stream `path` through blake3 in bounded memory.
fn blake3_file(path: &Path) -> Result<String> {
let mut f =
std::fs::File::open(path).with_context(|| format!("open {} to hash it", path.display()))?;
let mut h = blake3::Hasher::new();
let mut buf = vec![0u8; CHUNK];
loop {
let n = f
.read(&mut buf)
.with_context(|| format!("read {} while hashing", path.display()))?;
if n == 0 {
break;
}
h.update(&buf[..n]);
}
Ok(h.finalize().to_hex().to_string())
}
/// **What to acquire, where to put it, and how to know it is whole.**
#[derive(Clone, Debug)]
pub struct AcquireRequest {
pub source: Source,
/// The DURABLE destination. Never a temp dir and never `/tmp` (which is tmpfs — RAM —
/// on the boxes this runs on): a 176 GB transfer that lands in RAM is a transfer that
/// cannot land at all.
pub dest: PathBuf,
pub complete: Completeness,
/// Cooperative stop. This is the crate's existing
/// [`Shutdown`](crate::daemon::Shutdown) — the same flag a collector's run loop
/// already checks — rather than a second cancel type, so one `SIGTERM` stops the
/// download and the loop through one primitive.
pub cancel: Shutdown,
}
impl AcquireRequest {
/// An HTTP fetch to `dest`.
pub fn http(url: impl Into<String>, dest: impl Into<PathBuf>) -> Self {
Self {
source: Source::Http { url: url.into() },
dest: dest.into(),
complete: Completeness::Any,
cancel: Shutdown::new(),
}
}
/// A declared local seed linked to `dest`.
pub fn local_seed(declared: Vec<PathBuf>, dest: impl Into<PathBuf>) -> Self {
Self {
source: Source::LocalSeed { declared },
dest: dest.into(),
complete: Completeness::Any,
cancel: Shutdown::new(),
}
}
pub fn requiring(mut self, complete: Completeness) -> Self {
self.complete = complete;
self
}
pub fn cancelled_by(mut self, cancel: Shutdown) -> Self {
self.cancel = cancel;
self
}
/// The resume file — a sibling of `dest`, so the rename that finalises it is within
/// one filesystem and therefore atomic.
pub fn partial(&self) -> PathBuf {
let mut n = self.dest.file_name().unwrap_or_default().to_os_string();
n.push(".partial");
self.dest.with_file_name(n)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The sink
// ─────────────────────────────────────────────────────────────────────────────
/// **Where a transfer's progress and log go** — supplied by the supervisor, so a download
/// is observable by the thing nursing it.
///
/// This is the field whose absence produced the defect: `osm_katana::fixture` writes its
/// progress to an `indicatif` bar on its own stderr, so a second reader was written
/// downstream purely to infer the same number off the filesystem.
#[derive(Clone)]
pub struct AcquireSink {
/// `(done, total)` — `total` is `None` when the source did not say.
pub progress: Arc<dyn Fn(u64, Option<u64>) + Send + Sync>,
pub log: LogSink,
}
impl AcquireSink {
pub fn new(f: impl Fn(u64, Option<u64>) + Send + Sync + 'static, log: LogSink) -> Self {
Self {
progress: Arc::new(f),
log,
}
}
/// Discards progress; keeps the log.
pub fn quiet(log: LogSink) -> Self {
Self {
progress: Arc::new(|_, _| {}),
log,
}
}
/// A sink that records every progress callback — the seam that makes "did it report
/// progress at all" assertable.
pub fn recorder() -> (Self, Arc<std::sync::Mutex<Vec<(u64, Option<u64>)>>>) {
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let s = Arc::clone(&seen);
(
Self {
progress: Arc::new(move |d, t| {
s.lock().unwrap_or_else(|e| e.into_inner()).push((d, t))
}),
log: LogSink::noop(),
},
seen,
)
}
fn report(&self, done: u64, total: Option<u64>) {
(self.progress)(done, total);
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The result
// ─────────────────────────────────────────────────────────────────────────────
/// **What landed.** `resumed` and `transferred` are separate from `bytes` on purpose: a
/// caller that wants to know whether the network was touched cannot get that from a size.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Acquired {
pub path: PathBuf,
/// The artefact's total size.
pub bytes: u64,
/// Bytes moved by THIS call. `0` means the artefact was already there.
pub transferred: u64,
/// Whether this call continued an interrupted transfer.
pub resumed: bool,
/// The completeness declaration that was checked, for the provenance record.
pub checked: String,
}
/// **Ensure an artefact exists and is complete.** Idempotent: an already-landed artefact
/// returns without touching the network.
pub trait Acquire: Send + Sync {
fn acquire(&self, req: &AcquireRequest, sink: &AcquireSink) -> Result<Acquired>;
/// A short name for a log or a provenance `producer` field.
fn name(&self) -> &str;
}
/// The short-circuit every implementation owes its caller: if `dest` is already there and
/// satisfies the declaration, nothing happens.
fn already_landed(req: &AcquireRequest) -> Option<Acquired> {
if !req.dest.is_file() {
return None;
}
if req.complete.check(&req.dest).is_err() {
return None;
}
let bytes = std::fs::metadata(&req.dest).map(|m| m.len()).unwrap_or(0);
Some(Acquired {
path: req.dest.clone(),
bytes,
transferred: 0,
resumed: false,
checked: req.complete.label(),
})
}
/// fsync the file AND its directory, so the name is as durable as the bytes.
///
/// Same discipline as [`crate::state`]'s seal, and for the same reason: an artefact whose
/// bytes are on the platter under a name that is not is durable and unreachable.
fn fsync_file_and_parent(p: &Path) -> std::io::Result<()> {
std::fs::File::open(p)?.sync_all()?;
if let Some(dir) = p.parent() {
if let Ok(d) = std::fs::File::open(dir) {
let _ = d.sync_all();
}
}
Ok(())
}
// ─────────────────────────────────────────────────────────────────────────────
// LocalSeed — the operator's declared drop
// ─────────────────────────────────────────────────────────────────────────────
/// **An artefact the operator has already put on the box, at a path they DECLARED.**
///
/// # Why there is no `$HOME` fallback
///
/// There used to be one: `$HOME/planet-260427.osm.bz2`, tried as a first-class arm of the
/// planet resolver. It cost twice.
///
/// * It resolves against **whoever's home the process happens to run under**, so the
/// answer to "where does this service get its planet" depended on the invoking account
/// rather than on anything reviewable. That is the same class of mistake that put
/// 1 605 675 tiles in `/home/<login-user>/.korp` while the uninstall swept `/home/korp/.korp`
/// and reported success.
/// * Because it was tried first and always hit on the developer's box, **the transport
/// behind it was never exercised**. A fallback that always wins hides whatever it is a
/// fallback for.
///
/// A declared path is reviewable — it is in the config, an operator can read it, and a
/// service account that has no such file gets a refusal naming exactly which paths were
/// tried. So the operator drop stays; the guess does not.
#[derive(Clone, Debug, Default)]
pub struct LocalSeed {
/// Link rather than copy when possible — a 176 GB artefact must not be duplicated to
/// be used. A hardlink is tried first (it survives the seed being renamed), then a
/// symlink.
pub link: bool,
}
impl LocalSeed {
pub fn linking() -> Self {
Self { link: true }
}
pub fn copying() -> Self {
Self { link: false }
}
}
impl Acquire for LocalSeed {
fn acquire(&self, req: &AcquireRequest, sink: &AcquireSink) -> Result<Acquired> {
let Source::LocalSeed { declared } = &req.source else {
bail!(
"LocalSeed was handed a {:?} source — use the implementation that owns that \
transport",
req.source
);
};
if let Some(a) = already_landed(req) {
sink.log.line(format!(
"seed: {} is already present and satisfies {}",
a.path.display(),
a.checked
));
return Ok(a);
}
if declared.is_empty() {
bail!(
"no seed path is DECLARED, so there is nothing to link to {}. A declared path \
is reviewable; a $HOME search is not, and this refuses rather than guessing \
one from the account the process happens to run as",
req.dest.display()
);
}
let mut tried = Vec::new();
for cand in declared {
if !cand.is_file() {
tried.push(format!("{} (absent)", cand.display()));
continue;
}
if let Err(e) = req.complete.check(cand) {
tried.push(format!("{} ({e})", cand.display()));
continue;
}
if let Some(parent) = req.dest.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("mkdir {}", parent.display()))?;
}
let _ = std::fs::remove_file(&req.dest);
let how = self.place(cand, &req.dest)?;
let bytes = std::fs::metadata(cand)?.len();
sink.report(bytes, Some(bytes));
sink.log.line(format!(
"seed: {} → {} ({how}), {}",
cand.display(),
req.dest.display(),
req.complete.label()
));
return Ok(Acquired {
path: req.dest.clone(),
bytes,
transferred: 0,
resumed: false,
checked: req.complete.label(),
});
}
bail!(
"no declared seed satisfied {} — tried: {}",
req.complete.label(),
tried.join("; ")
)
}
fn name(&self) -> &str {
"local-seed"
}
}
impl LocalSeed {
fn place(&self, from: &Path, to: &Path) -> Result<&'static str> {
if self.link {
#[cfg(unix)]
{
if std::fs::hard_link(from, to).is_ok() {
return Ok("hardlink");
}
std::os::unix::fs::symlink(from, to).with_context(|| {
format!(
"neither hardlink nor symlink {} → {}",
from.display(),
to.display()
)
})?;
return Ok("symlink");
}
}
std::fs::copy(from, to)
.with_context(|| format!("copy {} → {}", from.display(), to.display()))?;
Ok("copy")
}
}
// ─────────────────────────────────────────────────────────────────────────────
// HttpRange — the resumable streaming fetch
// ─────────────────────────────────────────────────────────────────────────────
/// **A resumable HTTP fetch that never holds the body in memory.**
///
/// The body is streamed to `<dest>.partial` in [`CHUNK`] blocks, fsynced, and renamed
/// into place only after the completeness declaration passes. Three properties follow
/// from that order and each of them is one this replaces the absence of:
///
/// * **An interrupted transfer resumes.** The `.partial` is kept, its length becomes
/// `Range: bytes=<len>-`, and a `206` appends. A server that ignores the range and
/// answers `200` is detected and the file is restarted rather than corrupted by an
/// append of the whole body onto a prefix of itself.
/// * **A truncated file never appears under the final name.** `dest` exists only after
/// the check passes, so a later run cannot mistake half a download for a cache hit.
/// * **A stop is honoured mid-transfer** and leaves the `.partial`, which is what makes
/// the stop cheap: cancelling a 45-minute fetch costs the remainder, not the whole.
///
/// It carries no configuration. The client — user agent, connect timeout, and the absence
/// of a total timeout — is one per process (see [`shared_client`]), because a
/// `reqwest::blocking::Client` owns a tokio runtime and a per-request one is both wasteful
/// and a panic waiting for an async caller. A settable field here would advertise a knob
/// that the shared client does not honour.
#[derive(Clone, Copy, Debug, Default)]
pub struct HttpRange;
impl HttpRange {
pub fn new() -> Self {
Self::default()
}
}
/// **The process's ONE streaming HTTP client**, built off any async context and never
/// dropped.
///
/// Same hazard as [`crate::daemon::Fetcher::http`] and the same remedy, routed through the
/// same off-runtime builder: `reqwest::blocking::Client` owns a tokio runtime, and building
/// or dropping one while another runtime is current panics. A `OnceLock` means it is built
/// once and never dropped, so neither half of the pair can happen.
///
/// The configuration differs from the `Fetcher`'s on purpose. There is **no total
/// timeout**: a 176 GB transfer legitimately takes hours, and a total timeout is the classic
/// way such a fetch is killed at 90% for no reason. The guard that belongs on a large
/// transfer is a CONNECT timeout — a source that never answers must not hold the collector
/// forever — while a slow-but-moving stream is allowed to take as long as it takes.
fn shared_client() -> &'static reqwest::blocking::Client {
static CLIENT: std::sync::OnceLock<reqwest::blocking::Client> = std::sync::OnceLock::new();
CLIENT.get_or_init(|| {
crate::daemon::build_off_runtime(|| {
tls_roots_blocking(reqwest::blocking::Client::builder())
.user_agent(format!(
"korp-collectors/{} (+https://git.gunnar.rs/nordisk)",
env!("CARGO_PKG_VERSION")
))
.timeout(None)
.connect_timeout(std::time::Duration::from_secs(30))
.build()
.expect("build the streaming acquire HTTP client")
})
})
}
impl Acquire for HttpRange {
fn acquire(&self, req: &AcquireRequest, sink: &AcquireSink) -> Result<Acquired> {
let Source::Http { url } = &req.source else {
bail!(
"HttpRange was handed a {:?} source — use the implementation that owns that \
transport",
req.source
);
};
if let Some(a) = already_landed(req) {
sink.log.line(format!(
"acquire: {} is already present and satisfies {}",
a.path.display(),
a.checked
));
return Ok(a);
}
if let Some(parent) = req.dest.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("mkdir {}", parent.display()))?;
}
let partial = req.partial();
let have = std::fs::metadata(&partial).map(|m| m.len()).unwrap_or(0);
let client = shared_client();
let mut rb = client.get(url);
if have > 0 {
rb = rb.header(reqwest::header::RANGE, format!("bytes={have}-"));
}
let mut resp = rb.send().with_context(|| format!("GET {url}"))?;
if !resp.status().is_success() {
bail!("GET {url} → HTTP {}", resp.status());
}
// Did the server honour the range? 206 means yes; 200 with a range requested means
// it sent the whole body and appending would corrupt the file.
let resumed = have > 0 && resp.status() == reqwest::StatusCode::PARTIAL_CONTENT;
if have > 0 && !resumed {
sink.log.line(format!(
"acquire: {url} ignored Range: bytes={have}- (HTTP {}) — restarting the \
transfer rather than appending a whole body onto a prefix",
resp.status()
));
}
let start = if resumed { have } else { 0 };
let remaining = resp.content_length();
let total = remaining.map(|r| r + start);
let mut f = if resumed {
std::fs::OpenOptions::new()
.append(true)
.open(&partial)
.with_context(|| format!("reopen {} to resume", partial.display()))?
} else {
std::fs::File::create(&partial)
.with_context(|| format!("create {}", partial.display()))?
};
let mut done = start;
let mut transferred = 0u64;
let mut buf = vec![0u8; CHUNK];
sink.report(done, total);
loop {
if req.cancel.is_stopped() {
f.flush().ok();
drop(f);
bail!(
"acquire of {url} STOPPED at {done} bytes; {} is kept so the next attempt \
resumes from here rather than from zero",
partial.display()
);
}
let n = resp
.read(&mut buf)
.with_context(|| format!("read body of {url} at offset {done}"))?;
if n == 0 {
break;
}
f.write_all(&buf[..n])
.with_context(|| format!("write {} at offset {done}", partial.display()))?;
done += n as u64;
transferred += n as u64;
sink.report(done, total);
}
f.flush().context("flush the partial")?;
drop(f);
// Completeness is checked on the PARTIAL, before the rename. A file that fails
// must never wear the final name — that is the whole reason for the two-name
// dance, and checking after the rename would make the guard decorative.
if let Err(e) = req.complete.check(&partial) {
bail!(
"{e}. {} is kept for a resume; {} was NOT created, so no later run can mistake \
an incomplete transfer for a cache hit",
partial.display(),
req.dest.display()
);
}
fsync_file_and_parent(&partial)
.with_context(|| format!("fsync {} before finalising it", partial.display()))?;
std::fs::rename(&partial, &req.dest)
.with_context(|| format!("finalise {} → {}", partial.display(), req.dest.display()))?;
fsync_file_and_parent(&req.dest).ok();
let bytes = std::fs::metadata(&req.dest)?.len();
sink.log.line(format!(
"acquire: {url} → {} ({bytes} bytes, {transferred} transferred{}), {}",
req.dest.display(),
if resumed { ", resumed" } else { "" },
req.complete.label()
));
Ok(Acquired {
path: req.dest.clone(),
bytes,
transferred,
resumed,
checked: req.complete.label(),
})
}
fn name(&self) -> &str {
"http-range"
}
}
// ─────────────────────────────────────────────────────────────────────────────
// RED PROBES
// ─────────────────────────────────────────────────────────────────────────────
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots_blocking(b: reqwest::blocking::ClientBuilder) -> reqwest::blocking::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}
#[cfg(test)]
mod tests {
use super::*;
fn tmp() -> tempfile::TempDir {
tempfile::tempdir().unwrap()
}
// ── LocalSeed ────────────────────────────────────────────────────────────
#[test]
fn a_declared_seed_is_linked_not_copied_and_reports_its_size() {
let d = tmp();
let seed = d.path().join("declared.bin");
std::fs::write(&seed, vec![3u8; 4096]).unwrap();
let dest = d.path().join("cache").join("planet.bin");
let (sink, seen) = AcquireSink::recorder();
let got = LocalSeed::linking()
.acquire(
&AcquireRequest::local_seed(vec![seed.clone()], &dest)
.requiring(Completeness::MinBytes(4096)),
&sink,
)
.expect("link the declared seed");
assert_eq!(got.bytes, 4096);
assert_eq!(got.transferred, 0, "a local link transfers nothing");
assert!(dest.exists());
assert_eq!(std::fs::read(&dest).unwrap().len(), 4096);
assert!(
!seen.lock().unwrap().is_empty(),
"progress must be reported even for a seed"
);
// Idempotent: a second call is a no-op that still answers.
let again = LocalSeed::linking()
.acquire(
&AcquireRequest::local_seed(vec![seed], &dest)
.requiring(Completeness::MinBytes(4096)),
&sink,
)
.unwrap();
assert_eq!(again.bytes, 4096);
}
/// ★ **RED — `$HOME` is NOT a seed path.**
///
/// The file is in `$HOME` under the well-known name, exactly as the old resolver's
/// third arm expected, and nothing is declared. It must be refused. This is the probe
/// that would have failed while the `$HOME` arm existed, and it is why the torrent
/// branch went unexercised for months.
#[test]
fn a_seed_sitting_in_home_is_refused_when_no_path_is_declared() {
let d = tmp();
let fake_home = d.path().join("home");
std::fs::create_dir_all(&fake_home).unwrap();
std::fs::write(fake_home.join("planet-260427.osm.bz2"), vec![1u8; 2048]).unwrap();
let err = LocalSeed::linking()
.acquire(
&AcquireRequest::local_seed(vec![], d.path().join("planet.bin")),
&AcquireSink::quiet(LogSink::noop()),
)
.expect_err("an undeclared seed must be refused");
let msg = err.to_string();
assert!(
msg.contains("DECLARED"),
"the refusal must name the rule: {msg}"
);
assert!(
msg.contains("$HOME"),
"the refusal must say what it is refusing to do: {msg}"
);
assert!(
!d.path().join("planet.bin").exists(),
"a refused seed must create nothing"
);
}
/// ★ **RED — a declared seed that is too short is refused BY NAME**, and every path
/// tried is listed. A refusal that does not say which paths it looked at sends an
/// operator to guess.
#[test]
fn a_short_seed_is_refused_and_the_reason_lists_every_path_tried() {
let d = tmp();
let short = d.path().join("short.bin");
std::fs::write(&short, vec![0u8; 10]).unwrap();
let missing = d.path().join("nope.bin");
let err = LocalSeed::linking()
.acquire(
&AcquireRequest::local_seed(
vec![missing.clone(), short.clone()],
d.path().join("out.bin"),
)
.requiring(Completeness::MinBytes(1024)),
&AcquireSink::quiet(LogSink::noop()),
)
.expect_err("a 10-byte seed cannot satisfy ≥1024");
let msg = err.to_string();
assert!(msg.contains("nope.bin (absent)"), "{msg}");
assert!(msg.contains("short of the declared minimum 1024"), "{msg}");
assert!(!d.path().join("out.bin").exists());
}
// ── Completeness ─────────────────────────────────────────────────────────
/// ★ Every arm of the completeness enum must be able to say NO, and the digest arm
/// must agree with a digest computed independently.
#[test]
fn every_completeness_arm_can_fail_and_the_digest_arm_is_the_real_hash() {
let d = tmp();
let p = d.path().join("x.bin");
std::fs::write(&p, b"abcdefghij").unwrap();
assert!(Completeness::Any.check(&p).is_ok());
assert!(Completeness::MinBytes(10).check(&p).is_ok());
assert!(
Completeness::MinBytes(11).check(&p).is_err(),
"MinBytes must be able to fail"
);
assert!(Completeness::ExactBytes(10).check(&p).is_ok());
assert!(
Completeness::ExactBytes(9).check(&p).is_err(),
"ExactBytes must be able to fail"
);
let real = blake3::hash(b"abcdefghij").to_hex().to_string();
assert!(
Completeness::Blake3(real.clone()).check(&p).is_ok(),
"the streaming hash must match blake3 of the same bytes"
);
let wrong = Completeness::Blake3("00".repeat(32))
.check(&p)
.unwrap_err()
.to_string();
assert!(
wrong.contains(&real),
"the refusal must name the digest it got: {wrong}"
);
// An absent file is a failure for every arm, including `Any`.
let gone = d.path().join("gone.bin");
assert!(
Completeness::Any.check(&gone).is_err(),
"`Any` must still require the file to exist"
);
}
// ── HttpRange ────────────────────────────────────────────────────────────
/// A server that serves `body`, honours `Range:` when asked, and records every
/// request line so a test can assert what was actually asked for.
fn spawn_range_server(
body: Vec<u8>,
serve: usize,
honour_range: bool,
) -> (
String,
Arc<std::sync::Mutex<Vec<String>>>,
std::thread::JoinHandle<()>,
) {
use std::net::TcpListener;
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = l.local_addr().unwrap();
let seen: Arc<std::sync::Mutex<Vec<String>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
let s = Arc::clone(&seen);
let h = std::thread::spawn(move || {
for _ in 0..serve {
let Ok((mut st, _)) = l.accept() else { break };
let mut buf = [0u8; 2048];
let n = st.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]).to_string();
let from = req
.lines()
.find_map(|l| {
l.strip_prefix("range: ")
.or_else(|| l.strip_prefix("Range: "))
})
.and_then(|v| v.trim().strip_prefix("bytes="))
.and_then(|v| v.trim_end_matches('-').parse::<usize>().ok());
s.lock().unwrap().push(req);
match from {
Some(off) if honour_range && off < body.len() => {
let part = &body[off..];
let hdr = format!(
"HTTP/1.1 206 Partial Content\r\nContent-Length: {}\r\nContent-Range: bytes {}-{}/{}\r\nConnection: close\r\n\r\n",
part.len(),
off,
body.len() - 1,
body.len()
);
let _ = st.write_all(hdr.as_bytes());
let _ = st.write_all(part);
}
_ => {
let hdr = format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = st.write_all(hdr.as_bytes());
let _ = st.write_all(&body);
}
}
let _ = st.flush();
}
});
(format!("http://{addr}/artefact.bin"), seen, h)
}
#[test]
fn a_fetch_streams_to_a_partial_then_renames_and_reports_progress() {
let body: Vec<u8> = (0..300_000u32).map(|i| (i % 251) as u8).collect();
let (url, seen, h) = spawn_range_server(body.clone(), 1, true);
let d = tmp();
let dest = d.path().join("sub").join("artefact.bin");
let (sink, progress) = AcquireSink::recorder();
let got = HttpRange::new()
.acquire(
&AcquireRequest::http(&url, &dest).requiring(Completeness::ExactBytes(300_000)),
&sink,
)
.expect("fetch");
assert_eq!(got.bytes, 300_000);
assert_eq!(got.transferred, 300_000);
assert!(!got.resumed);
assert_eq!(
std::fs::read(&dest).unwrap(),
body,
"the bytes are the served bytes"
);
assert!(
!dest.with_file_name("artefact.bin.partial").exists(),
"the partial is gone"
);
let p = progress.lock().unwrap();
assert!(
p.len() >= 2,
"progress must be reported more than once: {p:?}"
);
assert_eq!(p.last().unwrap().0, 300_000);
assert_eq!(
p.last().unwrap().1,
Some(300_000),
"the total must be carried when the source gives one"
);
assert_eq!(seen.lock().unwrap().len(), 1);
h.join().unwrap();
}
/// ★ **RED — a truncated transfer must NOT create `dest`,** and the partial must be
/// kept. This is the property that makes a cache hit trustworthy: if a short file
/// could land under the final name, every later run would read it as complete.
#[test]
fn a_short_body_is_refused_leaves_no_dest_and_keeps_the_partial() {
let (url, _seen, h) = spawn_range_server(vec![9u8; 100], 1, true);
let d = tmp();
let dest = d.path().join("artefact.bin");
let err = HttpRange::new()
.acquire(
&AcquireRequest::http(&url, &dest).requiring(Completeness::MinBytes(1_000_000)),
&AcquireSink::quiet(LogSink::noop()),
)
.expect_err("100 bytes cannot satisfy ≥1000000");
let msg = err.to_string();
assert!(msg.contains("short of the declared minimum"), "{msg}");
assert!(
!dest.exists(),
"a refused transfer must NOT create the final name"
);
let partial = d.path().join("artefact.bin.partial");
assert!(partial.exists(), "the partial must be kept for a resume");
assert_eq!(std::fs::metadata(&partial).unwrap().len(), 100);
h.join().unwrap();
}
/// ★ **RED — an interrupted transfer RESUMES**: only the missing tail crosses the
/// wire, the `Range:` header says so, and the result is the whole artefact.
#[test]
fn an_interrupted_transfer_resumes_and_only_the_tail_is_transferred() {
let body: Vec<u8> = (0..200_000u32).map(|i| (i % 97) as u8).collect();
let d = tmp();
let dest = d.path().join("artefact.bin");
// Stage a half-finished transfer, exactly as a cancelled run leaves one.
std::fs::write(d.path().join("artefact.bin.partial"), &body[..80_000]).unwrap();
let (url, seen, h) = spawn_range_server(body.clone(), 1, true);
let got = HttpRange::new()
.acquire(
&AcquireRequest::http(&url, &dest).requiring(Completeness::ExactBytes(200_000)),
&AcquireSink::quiet(LogSink::noop()),
)
.expect("resume");
assert!(got.resumed, "the transfer must report that it resumed");
assert_eq!(got.bytes, 200_000);
assert_eq!(
got.transferred, 120_000,
"only the tail may cross the wire — a resume that re-sends the head is not a resume"
);
assert_eq!(std::fs::read(&dest).unwrap(), body);
assert!(
seen.lock().unwrap()[0].contains("bytes=80000-"),
"the request must carry the range: {}",
seen.lock().unwrap()[0]
);
h.join().unwrap();
}
/// ★ **RED — a server that IGNORES the range must not corrupt the file.** It answers
/// 200 with the whole body; appending that onto the existing prefix would produce a
/// file that is longer than the artefact and passes a `MinBytes` check.
#[test]
fn a_server_that_ignores_range_restarts_instead_of_appending() {
let body: Vec<u8> = (0..50_000u32).map(|i| (i % 13) as u8).collect();
let d = tmp();
let dest = d.path().join("artefact.bin");
std::fs::write(d.path().join("artefact.bin.partial"), &body[..20_000]).unwrap();
let (url, _seen, h) = spawn_range_server(body.clone(), 1, false);
let got = HttpRange::new()
.acquire(
&AcquireRequest::http(&url, &dest).requiring(Completeness::ExactBytes(50_000)),
&AcquireSink::quiet(LogSink::noop()),
)
.expect("restart");
assert!(!got.resumed, "a 200 answer is not a resume");
assert_eq!(
got.bytes, 50_000,
"an append would have produced 70000 bytes"
);
assert_eq!(std::fs::read(&dest).unwrap(), body);
h.join().unwrap();
}
/// ★ **RED — a stop mid-transfer is honoured, leaves no `dest`, and keeps the
/// partial** so the next attempt pays only the remainder.
#[test]
fn a_stop_mid_transfer_keeps_the_partial_and_creates_no_dest() {
// 8 MiB so the loop runs several chunks; the flag is already set, so the FIRST
// check trips and nothing is written.
let body = vec![4u8; 8 << 20];
let (url, _seen, h) = spawn_range_server(body, 1, true);
let d = tmp();
let dest = d.path().join("artefact.bin");
let stop = Shutdown::new();
stop.stop();
let err = HttpRange::new()
.acquire(
&AcquireRequest::http(&url, &dest).cancelled_by(stop),
&AcquireSink::quiet(LogSink::noop()),
)
.expect_err("a stopped acquire must fail, not silently succeed");
assert!(err.to_string().contains("STOPPED"), "{err}");
assert!(
!dest.exists(),
"a stopped transfer must not create the final name"
);
assert!(
d.path().join("artefact.bin.partial").exists(),
"the partial is the resume point"
);
h.join().unwrap();
}
/// An already-landed artefact costs NO request. The server is prepared to serve zero,
/// so a request would fail to connect — the counter is not the only proof.
#[test]
fn an_already_landed_artefact_touches_no_network() {
let d = tmp();
let dest = d.path().join("artefact.bin");
std::fs::write(&dest, vec![1u8; 4096]).unwrap();
let got = HttpRange::new()
.acquire(
// A URL nothing is listening on: reaching the network here is a failure.
&AcquireRequest::http("http://127.0.0.1:1/artefact.bin", &dest)
.requiring(Completeness::MinBytes(4096)),
&AcquireSink::quiet(LogSink::noop()),
)
.expect("a landed artefact must not be re-fetched");
assert_eq!(got.transferred, 0);
assert_eq!(got.bytes, 4096);
}
/// …and a landed artefact that does NOT satisfy the declaration is re-fetched rather
/// than accepted. Without this, the short-circuit above would be a way to make a bad
/// file permanent.
#[test]
fn a_landed_artefact_that_fails_its_declaration_is_refetched() {
let body = vec![7u8; 5000];
let (url, seen, h) = spawn_range_server(body.clone(), 1, true);
let d = tmp();
let dest = d.path().join("artefact.bin");
std::fs::write(&dest, vec![7u8; 10]).unwrap();
let got = HttpRange::new()
.acquire(
&AcquireRequest::http(&url, &dest).requiring(Completeness::ExactBytes(5000)),
&AcquireSink::quiet(LogSink::noop()),
)
.expect("refetch");
assert_eq!(got.bytes, 5000);
assert_eq!(
seen.lock().unwrap().len(),
1,
"the short artefact must have been re-fetched"
);
h.join().unwrap();
}
/// A source handed to the wrong implementation is refused, not silently mishandled.
#[test]
fn a_mismatched_source_is_refused_by_each_implementation() {
let d = tmp();
let e1 = HttpRange::new()
.acquire(
&AcquireRequest::local_seed(vec![], d.path().join("a")),
&AcquireSink::quiet(LogSink::noop()),
)
.unwrap_err()
.to_string();
assert!(e1.contains("LocalSeed"), "{e1}");
let e2 = LocalSeed::linking()
.acquire(
&AcquireRequest::http("http://x/y", d.path().join("b")),
&AcquireSink::quiet(LogSink::noop()),
)
.unwrap_err()
.to_string();
assert!(e2.contains("Http"), "{e2}");
}
}