src/harvest.rs
//! **The harvest — release-URL artefacts, off flaky links, into the pool.**
//!
//! # What is reused, and what is added
//!
//! The transfer engine is [`crate::download`]'s single attempt: `.partial` sibling,
//! `Range: bytes=N-` when one exists, `206`-vs-`200` handled so a server that ignores
//! the range restarts instead of appending a whole body onto a prefix, fsync before
//! rename. This module does not duplicate that arithmetic (one engine, not two).
//!
//! What is **added here** is the part one attempt cannot be:
//!
//! * **the retry loop.** One attempt on a link that drops at 60 % returns `Err` with
//! the `.partial` intact — correct, and useless on its own, because nobody re-enters
//! it. [`ReleaseFetch::acquire_with_resume`] is the loop that does, with exponential
//! backoff, and it is what makes a 4 GB ISO over a bad link finish.
//! * **a size floor checked BEFORE the rename.** Transport-level truncation is caught
//! by the engine (a body short of its `Content-Length` is a stream error), but a
//! *complete* 512-byte 404 page with an honest `Content-Length` is not. Without the
//! floor it wears the artefact's final name and every later run takes it for a
//! cache hit — permanently. See
//! [`tests::a_complete_but_undersized_body_is_refused_and_never_cached`].
//! * **content addressing on landing.** The bytes are hashed and taken into the
//! [`Pool`] by digest, and the artefact's name becomes a binding, not a filename.
//! * **a stop that is observed between attempts as well as during backoff.**
//! Cancelling a 45-minute fetch must cost the remainder, not the whole.
//!
//! # No daemon framework, no URLs
//!
//! Cancellation is `&dyn Fn() -> bool` and logging is `&dyn Fn(&str)` — the caller's
//! supervisor (a collector nursery, a CLI, a test) adapts its own types down to those
//! two closures, and this crate stays out of the daemon business. What to fetch is a
//! caller-supplied [`ReleaseArtifact`]; no URL is ever hardcoded here.
//!
//! Blocking, deliberately (the engine is `reqwest::blocking`). Call it from a
//! blocking-friendly context, e.g. `spawn_blocking`.
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use crate::coords::ReleaseArtifact;
use crate::download::{one_attempt, partial_path};
use crate::pool::{Digest, Pool};
/// What one artefact's fetch actually did.
///
/// `transferred` is the sum over every attempt, deliberately: it is the number that
/// tells a resumed transfer from a restarted one, and a per-attempt figure would hide
/// exactly the regression this module exists to prevent.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Fetched {
pub name: String,
pub digest: Digest,
/// The artefact's size.
pub bytes: u64,
/// Bytes pulled over the network by this call, across all attempts. `0` means the
/// staging cache already held it.
pub transferred: u64,
/// How many attempts it took. `1` on a clean link, `0` on a staging-cache hit.
pub attempts: u32,
/// Whether any attempt continued an earlier `.partial`.
pub resumed: bool,
/// `true` when the pool already held these bytes under this digest.
pub deduped: bool,
/// Where it landed under a human name — the download cache, hardlinked to the blob.
pub staged_at: PathBuf,
}
/// The totals an [`ReleaseFetch::acquire_with_resume`] run produced.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Landed {
/// Size of the landed file.
pub bytes: u64,
pub attempts: u32,
/// Summed over attempts, including bytes moved by attempts that then failed.
pub transferred: u64,
pub resumed: bool,
}
/// The release-URL fetcher: the retry/backoff/cancel loop.
#[derive(Clone, Debug)]
pub struct ReleaseFetch {
/// How many attempts before giving up on this pass. A supervised caller is
/// restarted anyway, so this bounds one pass, not the effort.
pub attempts: u32,
/// First delay between attempts; doubles, capped at [`ReleaseFetch::backoff_max`].
pub backoff_min: Duration,
pub backoff_max: Duration,
}
impl Default for ReleaseFetch {
fn default() -> Self {
Self {
attempts: 5,
backoff_min: Duration::from_millis(500),
backoff_max: Duration::from_secs(60),
}
}
}
impl ReleaseFetch {
/// An immediate-retry fetcher — for tests, where a real backoff is dead time and a
/// zero one still exercises the loop.
pub fn eager(attempts: u32) -> Self {
Self {
attempts,
backoff_min: Duration::ZERO,
backoff_max: Duration::ZERO,
}
}
/// **Fetch `art` into `staging`, retrying until it lands, then take it into `pool`.**
pub fn fetch_into_pool(
&self,
art: &ReleaseArtifact,
pool: &Pool,
staging: &Path,
log: &dyn Fn(&str),
cancelled: &dyn Fn() -> bool,
) -> Result<Fetched> {
let dest = staging.join(&art.name);
let landed = self.acquire_with_resume(art, &dest, log, cancelled)?;
// Hash the bytes that actually landed, and check them against a published digest
// when the upstream offers one. A dormant check that is WIRED is one config line
// from working; one that is absent is a rewrite.
let digest = Digest::of_file(&dest)?;
if let Some(want) = &art.expect_sha256 {
let want = Digest::parse(want)
.with_context(|| format!("`{}` declares an unparseable expect_sha256", art.name))?;
if want != digest {
// The file is deliberately left where it is rather than deleted: an
// artefact whose bytes are wrong is evidence, and a fetcher that erases
// its own evidence makes the next run look like the first.
bail!(
"{} hashes sha256:{digest} but its pin says sha256:{want} — the bytes at {} \
are not the bytes that were asked for. {} is left in place for inspection",
art.name,
art.url,
dest.display()
);
}
}
if landed.bytes < art.min_bytes {
bail!(
"{} landed {} bytes, under its declared floor of {} — INCOMPLETE",
art.name,
landed.bytes,
art.min_bytes
);
}
let ingested = pool.ingest_file(&dest)?;
debug_assert_eq!(ingested.digest, digest);
pool.bind(&art.name, &ingested.digest, &art.url)?;
Ok(Fetched {
name: art.name.clone(),
digest: ingested.digest,
bytes: landed.bytes,
transferred: landed.transferred,
attempts: landed.attempts,
resumed: landed.resumed,
deduped: !ingested.stored,
staged_at: dest,
})
}
/// **The retry loop.** One [`crate::download`] attempt per iteration; the `.partial`
/// it leaves behind is what makes the next iteration a resume rather than a restart.
///
/// The size floor runs on the `.partial` BEFORE the rename, so a complete-but-short
/// body never wears the final name — and it is discarded rather than kept, because a
/// complete wrong body is poison to resume onto, unlike an interrupted right one.
///
/// Returns the totals across attempts.
pub fn acquire_with_resume(
&self,
art: &ReleaseArtifact,
dest: &Path,
log: &dyn Fn(&str),
cancelled: &dyn Fn() -> bool,
) -> Result<Landed> {
// Already landed and plausible? Then this call is a no-op, not a re-transfer.
// (The pin, when there is one, is re-checked by the caller on every pass.)
if dest.exists() {
let len = std::fs::metadata(dest)?.len();
if len >= art.min_bytes {
return Ok(Landed {
bytes: len,
attempts: 0,
transferred: 0,
resumed: false,
});
}
// A too-small file at the final name is stale, not in-flight: refetch.
std::fs::remove_file(dest)
.with_context(|| format!("remove stale {}", dest.display()))?;
}
if let Some(parent) = dest.parent()
&& !parent.as_os_str().is_empty()
{
std::fs::create_dir_all(parent)
.with_context(|| format!("create staging dir {}", parent.display()))?;
}
let client = crate::http_tls::tls_roots_blocking(reqwest::blocking::Client::builder())
.connect_timeout(Duration::from_secs(30))
// No whole-transfer timeout: a multi-GB ISO on a slow link is the normal
// case, and a dead link surfaces as a read error anyway.
.timeout(None)
.build()
.context("build HTTP client")?;
let partial = partial_path(dest);
let attempts = self.attempts.max(1);
let mut transferred = 0u64;
let mut resumed = false;
let mut delay = self.backoff_min;
let mut last: Option<anyhow::Error> = None;
for attempt in 1..=attempts {
if cancelled() {
bail!(
"fetch of {} STOPPED before attempt {attempt}; {} is kept so the next run \
resumes from there",
art.name,
partial.display()
);
}
let pre = partial_len(&partial);
let outcome: Result<u64> = match one_attempt(&client, &art.url, &partial, &mut resumed)
{
Ok(written) => {
transferred += written;
let have = partial_len(&partial);
if have >= art.min_bytes {
std::fs::rename(&partial, dest)
.with_context(|| format!("rename partial into {}", dest.display()))?;
return Ok(Landed {
bytes: have,
attempts: attempt,
transferred,
resumed,
});
}
// A COMPLETE body under the floor is a WRONG body (the 404 page
// where an ISO belongs), not an interrupted one. Discard it: a
// later attempt must not resume onto poison, and a later run must
// not mistake it for a cache hit.
std::fs::remove_file(&partial).ok();
Err(anyhow::anyhow!(
"{} landed {have} bytes, under its declared floor of {} — INCOMPLETE",
art.name,
art.min_bytes
))
}
Err(e) => {
// Count what this attempt DID move before it died. Without this the
// caller cannot tell a resume from a restart, which is the one
// property the whole module exists to hold.
transferred += partial_len(&partial).saturating_sub(pre);
Err(e)
}
};
let e = outcome.unwrap_err();
if cancelled() {
return Err(e.context(format!(
"fetch of {} stopped during attempt {attempt}",
art.name
)));
}
log(&format!(
"skidbladnir::harvest: {} attempt {attempt}/{attempts} failed: {e:#} — {} holds \
{} bytes, retrying in {delay:?}",
art.name,
partial.display(),
partial_len(&partial)
));
last = Some(e);
// Sleep in slices so a stop during a 60 s backoff is observed in 100 ms.
let until = std::time::Instant::now() + delay;
while std::time::Instant::now() < until {
if cancelled() {
bail!("fetch of {} STOPPED during backoff", art.name);
}
std::thread::sleep(
Duration::from_millis(50).min(until - std::time::Instant::now()),
);
}
delay = (delay * 2).min(self.backoff_max.max(self.backoff_min));
}
let held = partial_len(&partial);
Err(anyhow::anyhow!(
"gave up on {} after {attempts} attempt(s); {} holds {held} bytes and the NEXT run \
resumes from there rather than from zero. Last error: {}",
art.name,
partial.display(),
last.map(|e| format!("{e:#}"))
.unwrap_or_else(|| "none".into())
))
}
}
fn partial_len(p: &Path) -> u64 {
std::fs::metadata(p).map(|m| m.len()).unwrap_or(0)
}
/// A no-op log sink and a never-tripped cancel, for callers with neither.
pub fn quiet() -> (impl Fn(&str), impl Fn() -> bool) {
(|_: &str| {}, || false)
}
// ─────────────────────────────────────────────────────────────────────────────
// RED PROBES — wire-side truth, asserted on a real socket
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use crate::testserv::{Behaviour, TestServer};
/// A body big enough that a truncation is unambiguous and small enough that the test
/// is instant. Pseudo-random so a resumed suffix cannot accidentally hash like a
/// restarted whole.
fn payload(n: usize) -> Vec<u8> {
let mut v = Vec::with_capacity(n);
let mut x: u64 = 0x9E3779B97F4A7C15;
while v.len() < n {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
v.extend_from_slice(&x.to_le_bytes());
}
v.truncate(n);
v
}
fn art(url: &str, min_bytes: u64) -> ReleaseArtifact {
ReleaseArtifact {
name: "talos-metal-amd64.iso".into(),
url: url.into(),
expect_sha256: None,
min_bytes,
}
}
fn no_log() -> impl Fn(&str) {
|_: &str| {}
}
fn never() -> impl Fn() -> bool {
|| false
}
/// ★★ **THE test.** A real HTTP server, a transfer killed mid-stream, a resume, and
/// the assertion is on the **reassembled bytes' sha256** — not on a flag, not on a
/// state round-trip.
///
/// Four independent things are asserted, and each one can fail on its own:
///
/// 1. the landed file's sha256 equals the whole payload's — bytes, not bookkeeping;
/// 2. the server saw `Range: bytes=40000-` on request 2 — the client *asked* to
/// resume, which a lucky refetch would not show;
/// 3. the server wrote the payload **exactly once** across both requests — a restart
/// would have written 40 000 + 100 000, so this is the assertion that catches a
/// resume that silently became a refetch;
/// 4. the blob is in the pool under that digest and the binding resolves to it.
///
/// **SEEN RED 2026-08-07 — probe B** (`let _ = std::fs::remove_file(…partial…)`
/// added to the retry loop's error arm, so every attempt starts from zero):
/// `attempt 2 continued the .partial`. Three sibling tests went red with it. That is
/// the probe that matters, because it breaks resume *specifically* while leaving the
/// transfer able to complete.
#[test]
fn an_interrupted_transfer_resumes_and_the_reassembled_bytes_hash_correctly() {
let body = payload(100_000);
let want = Digest::of_bytes(&body);
let srv = TestServer::start(
body.clone(),
// Attempt 1 dies after 40 000 of the 100 000 promised bytes; attempt 2 is
// answered honestly.
vec![Behaviour::Truncate(40_000), Behaviour::Honour],
);
let d = tempfile::tempdir().unwrap();
let pool = Pool::open(d.path().join("pool")).unwrap();
let staging = d.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
let a = art(&srv.url, 100_000);
let got = ReleaseFetch::eager(4)
.fetch_into_pool(&a, &pool, &staging, &no_log(), &never())
.expect("the retry loop must carry an interrupted transfer to completion");
// (1) APPLIED OUTPUT — the bytes on disk.
assert_eq!(
got.digest, want,
"the reassembled file must hash to the payload"
);
assert_eq!(got.bytes, 100_000);
assert_eq!(
std::fs::read(&got.staged_at).unwrap(),
body,
"byte-for-byte, not just the same length"
);
assert!(got.resumed, "attempt 2 continued the .partial");
assert_eq!(got.attempts, 2, "one failure, one success");
// (2) The client ASKED to resume — from exactly where it stopped.
let seen = srv.seen();
assert_eq!(seen.len(), 2, "two requests: {seen:?}");
assert_eq!(seen[0].range, None, "the first request has no Range");
assert_eq!(
seen[1].range.as_deref(),
Some("bytes=40000-"),
"the second request resumes from the 40 000 bytes already on disk: {seen:?}"
);
assert_eq!(seen[1].status, 206);
// (3) THE anti-identity assertion. A refetch-from-zero would also produce a
// correct file and a `resumed` flag could still be true; only the wire count
// separates them. 40 000 + 60 000 = 100 000, exactly the payload, once.
assert_eq!(
srv.bytes_written(),
100_000,
"the payload crossed the wire exactly once — a restart would be 140 000: {seen:?}"
);
assert_eq!(
got.transferred, 100_000,
"and the client's own count agrees with the wire"
);
// (4) It is in the pool, by content, under its name.
assert!(pool.has(&want));
assert_eq!(pool.read(&want).unwrap(), body);
let b = pool.resolve(&a.name).unwrap().expect("bound");
assert_eq!(b.digest, want);
assert_eq!(b.len, 100_000);
assert_eq!(b.source, srv.url);
assert!(pool.verify().unwrap().is_empty());
}
/// ★ **RED — a server that IGNORES `Range` must restart, not corrupt.**
///
/// The failure produces a file that is the right size in nobody's arithmetic:
/// 40 000 bytes of prefix with 100 000 bytes appended. The assertion is the digest
/// and the length, so a corrupting append fails both.
///
/// **SEEN RED 2026-08-07 — probe D** (request 2's script entry changed from
/// `IgnoreRange` to `Honour`, so the server resumes instead of ignoring):
/// `a 200 answer is a restart, not a resume`. So the test really does discriminate
/// the `200` path from the `206` path rather than passing on both.
#[test]
fn a_server_that_ignores_range_restarts_rather_than_corrupting_the_file() {
let body = payload(100_000);
let want = Digest::of_bytes(&body);
let srv = TestServer::start(
body.clone(),
vec![
Behaviour::Truncate(40_000),
// A `200` carrying the WHOLE body despite the `Range` header.
Behaviour::IgnoreRange,
],
);
let d = tempfile::tempdir().unwrap();
let pool = Pool::open(d.path().join("pool")).unwrap();
let staging = d.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
let got = ReleaseFetch::eager(4)
.fetch_into_pool(
&art(&srv.url, 100_000),
&pool,
&staging,
&no_log(),
&never(),
)
.expect("an ignored Range must still land a correct file");
assert_eq!(
got.bytes, 100_000,
"140 000 here would mean the whole body was appended onto the 40 000-byte prefix"
);
assert_eq!(got.digest, want);
assert_eq!(std::fs::read(&got.staged_at).unwrap(), body);
assert!(!got.resumed, "a 200 answer is a restart, not a resume");
let seen = srv.seen();
assert_eq!(
seen[1].range.as_deref(),
Some("bytes=40000-"),
"the client did ask; the server chose to ignore it"
);
assert_eq!(seen[1].status, 200);
// The honest cost of a server that will not resume: the prefix travelled twice.
assert_eq!(srv.bytes_written(), 140_000);
}
/// ★ **RED — a short body never wears the final name.**
///
/// Every attempt is truncated, so the fetch must fail; what matters is the state it
/// leaves. `dest` must NOT exist (a later run would take it for a cache hit and
/// never refetch), the `.partial` MUST exist (so the next run resumes), and the
/// error must say where the bytes are.
///
/// **SEEN RED 2026-08-07 — probe B**: `the .partial must survive for the resume`.
/// With the accumulation broken, `held` is 5 000 rather than 20 000, which is the
/// difference between a fetch that makes progress across restarts and one that
/// does not.
#[test]
fn a_transfer_that_never_completes_leaves_a_partial_and_no_final_name() {
let body = payload(100_000);
let srv = TestServer::start(
body,
vec![
Behaviour::Truncate(10_000),
Behaviour::Truncate(5_000),
Behaviour::Truncate(5_000),
],
);
let d = tempfile::tempdir().unwrap();
let pool = Pool::open(d.path().join("pool")).unwrap();
let staging = d.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
let a = art(&srv.url, 100_000);
let e = ReleaseFetch::eager(3)
.fetch_into_pool(&a, &pool, &staging, &no_log(), &never())
.expect_err("three truncated attempts cannot produce a complete artefact");
let dest = staging.join(&a.name);
assert!(
!dest.is_file(),
"an incomplete transfer must NOT wear the final name — {} exists, so every later \
run would take it for a cache hit",
dest.display()
);
let partial = staging.join(format!("{}.partial", a.name));
assert!(
partial.is_file(),
"the .partial must survive for the resume"
);
let held = std::fs::metadata(&partial).unwrap().len();
assert_eq!(
held, 20_000,
"10 000 + 5 000 + 5 000 accumulated across attempts — this is the number that \
would be 5 000 if each attempt restarted from zero"
);
let m = format!("{e:#}");
assert!(m.contains("resumes from there"), "{m}");
assert!(m.contains("20000"), "the error names the bytes held: {m}");
// Nothing entered the pool, and nothing is bound. A half-download must not be
// reachable by name.
assert!(pool.digests().unwrap().is_empty());
assert!(pool.resolve(&a.name).unwrap().is_none());
}
/// ★ **RED — a COMPLETE body that is simply too small is refused and never cached.**
///
/// This test exists because a red probe once came back green: swapping the floor out
/// left every truncation test green, because the HTTP stack rejects a body short of
/// its own `Content-Length` before any floor is consulted. The floor's actual job is
/// this case: an upstream that answers a perfectly well-formed 512-byte error page
/// where a 100 MB ISO should be.
///
/// The failure it prevents is permanent, which is why it is worth its own test: with
/// no floor the page is renamed to `talos-metal-amd64.iso`, the staging-cache
/// short-circuit fires on every later run, and the bundle ships an HTML file as an
/// ISO until someone deletes it by hand.
#[test]
fn a_complete_but_undersized_body_is_refused_and_never_cached() {
// 550 bytes, served whole, with an honest Content-Length. Nothing about the
// transfer is wrong — only the size.
let page = b"<html><head><title>404 Not Found</title></head></html>".repeat(10);
assert!(page.len() < 1000);
let srv = TestServer::start(page.clone(), vec![Behaviour::Honour, Behaviour::Honour]);
let d = tempfile::tempdir().unwrap();
let pool = Pool::open(d.path().join("pool")).unwrap();
let staging = d.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
let a = art(&srv.url, 32 * 1024 * 1024);
let e = ReleaseFetch::eager(2)
.fetch_into_pool(&a, &pool, &staging, &no_log(), &never())
.expect_err("a 512-byte answer where an ISO belongs must refuse");
assert!(
!staging.join(&a.name).is_file(),
"the error page must NOT wear the artefact's name — that is the cache-hit \
that never expires"
);
assert!(format!("{e:#}").contains("INCOMPLETE"), "{e:#}");
assert!(
pool.digests().unwrap().is_empty(),
"and nothing entered the pool"
);
assert!(pool.resolve(&a.name).unwrap().is_none());
}
/// A second fetch of an artefact already staged touches the network **zero** times.
/// Asserted on the server's own request count, which the client cannot fake — and
/// the delta since a snapshot cut between the passes is genuinely empty, which is
/// the arithmetic the whole design rests on.
///
/// **SEEN RED 2026-08-07 — probe C** (`std::fs::remove_file(&dest)` added after the
/// pool ingest, so the staging cache is destroyed): `no bytes moved, left: 50000,
/// right: 0`, and the server saw a second request.
#[test]
fn an_already_landed_artefact_costs_no_request() {
let body = payload(50_000);
let srv = TestServer::start(body.clone(), vec![Behaviour::Honour]);
let d = tempfile::tempdir().unwrap();
let pool = Pool::open(d.path().join("pool")).unwrap();
let staging = d.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
let a = art(&srv.url, 50_000);
let f = ReleaseFetch::eager(2);
let first = f
.fetch_into_pool(&a, &pool, &staging, &no_log(), &never())
.unwrap();
assert_eq!(first.transferred, 50_000);
assert!(!first.deduped, "the first ingest stores");
assert_eq!(srv.seen().len(), 1);
let second = f
.fetch_into_pool(&a, &pool, &staging, &no_log(), &never())
.unwrap();
assert_eq!(second.transferred, 0, "no bytes moved");
assert!(second.deduped, "the pool already held this content");
assert_eq!(second.digest, first.digest);
assert_eq!(
srv.seen().len(),
1,
"and the SERVER saw no second request — the cache is real, not reported"
);
assert_eq!(pool.digests().unwrap().len(), 1, "still one blob");
}
/// ★ **RED — a stop during backoff is observed, and the `.partial` survives it.**
/// A cancelled 45-minute fetch must cost the remainder, not the whole.
#[test]
fn a_stop_between_attempts_is_observed_and_keeps_the_partial() {
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
let body = payload(100_000);
let srv = TestServer::start(body, vec![Behaviour::Truncate(30_000)]);
let d = tempfile::tempdir().unwrap();
let staging = d.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
let a = art(&srv.url, 100_000);
let stop = Arc::new(AtomicBool::new(false));
let s2 = Arc::clone(&stop);
// Stop shortly after the first attempt dies, while the loop is in its backoff.
// ROOT-LAW-#0: through gatling's ONE sanctioned spawn.
let trip = gatling::background::Job::spawn(move || {
std::thread::sleep(Duration::from_millis(120));
s2.store(true, Ordering::SeqCst);
});
let f = ReleaseFetch {
attempts: 10,
backoff_min: Duration::from_secs(5),
backoff_max: Duration::from_secs(5),
};
let t = std::time::Instant::now();
let e = f
.acquire_with_resume(&a, &staging.join(&a.name), &no_log(), &|| {
stop.load(Ordering::SeqCst)
})
.expect_err("a stopped fetch must not report success");
assert!(
t.elapsed() < Duration::from_secs(4),
"the stop must be seen inside the 5 s backoff, not after it — took {:?}",
t.elapsed()
);
assert!(format!("{e:#}").contains("STOPPED"), "{e:#}");
let partial = staging.join(format!("{}.partial", a.name));
assert_eq!(
std::fs::metadata(&partial).unwrap().len(),
30_000,
"the 30 000 bytes already paid for are kept"
);
let _ = trip.join();
}
/// ★ **RED — a landed file whose bytes contradict a published digest is REFUSED**,
/// and the file is kept as evidence rather than erased.
#[test]
fn a_digest_mismatch_refuses_and_keeps_the_evidence() {
let body = payload(20_000);
let srv = TestServer::start(body.clone(), vec![Behaviour::Honour]);
let d = tempfile::tempdir().unwrap();
let pool = Pool::open(d.path().join("pool")).unwrap();
let staging = d.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
let mut a = art(&srv.url, 20_000);
a.expect_sha256 = Some(Digest::of_bytes(b"a different artefact entirely").to_string());
let e = ReleaseFetch::eager(2)
.fetch_into_pool(&a, &pool, &staging, &no_log(), &never())
.expect_err("bytes that contradict the pin must refuse");
let m = format!("{e:#}");
assert!(m.contains("are not the bytes that were asked for"), "{m}");
assert!(
staging.join(&a.name).is_file(),
"the wrong file is left for inspection"
);
assert!(
pool.digests().unwrap().is_empty(),
"and it did NOT enter the pool"
);
assert!(pool.resolve(&a.name).unwrap().is_none());
// …and the matching pin passes, or the red above proves nothing.
let mut ok = art(&srv.url, 20_000);
ok.expect_sha256 = Some(Digest::of_bytes(&body).to_string());
ok.name = "same-bytes-right-pin".into();
let got = ReleaseFetch::eager(2)
.fetch_into_pool(&ok, &pool, &staging, &no_log(), &never())
.expect("a correct pin must pass");
assert_eq!(got.digest, Digest::of_bytes(&body));
}
}