src/download.rs
//! Resumable HTTP download — the CONNECTED-side transfer utility.
//!
//! [`crate::fetch`] is the airgap half: it does **zero network IO** by design,
//! because the target never reaches out. This module is the other half, for the
//! connected side that ASSEMBLES a bundle (the collector pattern): pull an
//! upstream artefact to disk so it survives flaky links, and never let a short
//! read wear the final filename.
//!
//! The shape is `dlFile`'s from the Go airgap-kubernetes repo — the one
//! capability that repo had and this crate did not (its content *delta* is
//! already here, in [`crate::archive::Archive::delta`]) — plus the retry loop
//! `dlFile` gives up without:
//!
//! - the in-flight bytes live in a `<dest>.partial` sibling, never at `dest`;
//! - an interrupted transfer resumes with `Range: bytes=N-`;
//! - a server that answers a Range request with `200` (ignoring the range)
//! RESTARTS the partial cleanly instead of appending a whole body onto a
//! prefix — the corruption `dlFile` guards against, kept here;
//! - the integrity check runs on the PARTIAL, before the rename, so a wrong
//! or truncated file never becomes `dest`;
//! - the rename is atomic (same directory), and the file is fsynced first.
//!
//! **No URLs live here.** What to download and from where is the CALLER's
//! knowledge (talos, vault, tofu pins belong to the consumer, e.g. moln);
//! this module owns only *how* bytes move and when they are trusted.
use std::fs::{self, OpenOptions};
use std::io::{Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use crate::archive::sha256_file;
/// How hard to try. No URL, no destination — those are per-call.
#[derive(Debug, Clone)]
pub struct DownloadPolicy {
/// Total attempts per call (first try included). At least 1.
pub attempts: u32,
/// Flat pause between attempts. The caller's supervisor owns real backoff
/// schedules; this is only the courtesy gap between consecutive tries.
pub retry_pause: Duration,
/// TCP connect timeout. There is deliberately NO whole-transfer timeout:
/// a multi-hundred-MiB ISO on a slow link is the normal case, not a hang.
pub connect_timeout: Duration,
}
impl Default for DownloadPolicy {
fn default() -> Self {
DownloadPolicy {
attempts: 3,
retry_pause: Duration::from_secs(2),
connect_timeout: Duration::from_secs(30),
}
}
}
/// What a completed download attests.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Downloaded {
/// The final path — `dest` as given, now carrying the verified bytes.
pub path: PathBuf,
/// Size on disk after completion.
pub bytes: u64,
/// SHA-256 of the landed file (always computed, matched when expected).
pub sha256: String,
/// `true` when at least one attempt continued a previous partial with a
/// `Range` request (either from an earlier call or an earlier attempt).
pub resumed: bool,
/// Network attempts actually spent. `0` = `dest` already existed and
/// matched `expected_sha256`, so no request was made.
pub attempts: u32,
}
/// The `.partial` sibling where in-flight bytes live until they verify.
pub fn partial_path(dest: &Path) -> PathBuf {
let mut name = dest.file_name().unwrap_or_default().to_os_string();
name.push(".partial");
dest.with_file_name(name)
}
/// Download `url` to `dest`, resumably. See the module doc for the contract.
///
/// `expected_sha256` (lowercase hex) is the out-of-band pin: when given, the
/// partial must hash to it before it may become `dest` (a mismatch discards
/// the partial and counts as a failed attempt — poisoned bytes are never
/// resumed onto). When `None`, only transport-level completion is checked.
///
/// Idempotent: if `dest` already exists and matches the pin (or no pin was
/// given), no request is made and `attempts` is 0.
pub fn download_resumable(
url: &str,
dest: &Path,
expected_sha256: Option<&str>,
policy: &DownloadPolicy,
) -> Result<Downloaded> {
// Already landed? Then this call is a no-op, not a re-transfer.
if dest.exists() {
let have = sha256_file(dest)?;
match expected_sha256 {
Some(want) if have != want => {
// A wrong file at the final name is stale, not in-flight:
// remove it and transfer fresh.
fs::remove_file(dest)
.with_context(|| format!("remove stale {}", dest.display()))?;
}
_ => {
let bytes = fs::metadata(dest)?.len();
return Ok(Downloaded {
path: dest.to_path_buf(),
bytes,
sha256: have,
resumed: false,
attempts: 0,
});
}
}
}
if let Some(parent) = dest.parent()
&& !parent.as_os_str().is_empty()
{
fs::create_dir_all(parent)
.with_context(|| format!("create download dir {}", parent.display()))?;
}
let client = crate::http_tls::tls_roots_blocking(reqwest::blocking::Client::builder())
.connect_timeout(policy.connect_timeout)
// No whole-transfer timeout: big artefacts on slow links are the
// normal case here, and a dead link surfaces as a read error anyway.
.timeout(None)
.build()
.context("build HTTP client")?;
let partial = partial_path(dest);
let attempts = policy.attempts.max(1);
let mut resumed = false;
let mut last_err: Option<anyhow::Error> = None;
for attempt in 1..=attempts {
if attempt > 1 {
std::thread::sleep(policy.retry_pause);
}
match one_attempt(&client, url, &partial, &mut resumed) {
Ok(_written) => {
// Integrity runs on the PARTIAL: a wrong file never wears `dest`.
let have = sha256_file(&partial)?;
if let Some(want) = expected_sha256
&& have != want
{
// Poisoned bytes must not be resumed onto — discard.
fs::remove_file(&partial).ok();
last_err = Some(anyhow::anyhow!(
"sha256 mismatch for {url}: got {have}, want {want}"
));
crate::surface_status!(
"skidbladnir/download",
"download_resumable",
false,
"attempt {attempt}/{attempts}: digest mismatch, partial discarded"
);
continue;
}
let bytes = fs::metadata(&partial)?.len();
fs::rename(&partial, dest)
.with_context(|| format!("rename partial into {}", dest.display()))?;
crate::surface_status!(
"skidbladnir/download",
"download_resumable",
true,
"{url} -> {} ({bytes} B, attempt {attempt}/{attempts}, resumed={resumed})",
dest.display()
);
return Ok(Downloaded {
path: dest.to_path_buf(),
bytes,
sha256: have,
resumed,
attempts: attempt,
});
}
Err(e) => {
// The partial stays: the next attempt resumes from its length.
last_err = Some(e);
}
}
}
let err = last_err.unwrap_or_else(|| anyhow::anyhow!("no attempts were made"));
crate::surface_status!(
"skidbladnir/download",
"download_resumable",
false,
"{url} failed after {attempts} attempts: {err:#}"
);
Err(err.context(format!("download {url} failed after {attempts} attempts")))
}
/// One transfer attempt: open/extend the partial, issue the (ranged) GET,
/// stream the body, fsync. Leaves the partial in place on error so the next
/// attempt can resume from its byte count. Returns the bytes streamed by THIS
/// attempt (which, on a 200-restart, can exceed the partial's growth).
///
/// `pub(crate)` — [`crate::harvest`] drives this same engine under its own
/// retry/backoff/cancel loop rather than duplicating the range arithmetic.
pub(crate) fn one_attempt(
client: &reqwest::blocking::Client,
url: &str,
partial: &Path,
resumed: &mut bool,
) -> Result<u64> {
let offset = fs::metadata(partial).map(|m| m.len()).unwrap_or(0);
let mut req = client.get(url);
if offset > 0 {
req = req.header(reqwest::header::RANGE, format!("bytes={offset}-"));
}
let mut resp = req.send().with_context(|| format!("GET {url}"))?;
let status = resp.status();
// Not truncated on open: a 206 resumes after the partial's bytes; the 200
// arm below empties it itself.
let mut file = OpenOptions::new()
.create(true)
.append(false)
.truncate(false)
.write(true)
.open(partial)
.with_context(|| format!("open partial {}", partial.display()))?;
match (status.as_u16(), offset) {
// Fresh body. If we asked for a range and got 200, the server ignored
// it: RESTART from zero rather than appending a whole body to a prefix.
(200, n) => {
if n > 0 {
file.set_len(0)?;
}
file.seek(SeekFrom::Start(0))?;
}
// The server honoured the range: append after what we have.
(206, n) if n > 0 => {
*resumed = true;
file.seek(SeekFrom::Start(n))?;
}
// 206 without a Range request, or 416 (our partial confused the
// server — e.g. it already covers the file): restart clean.
(206, _) | (416, _) => {
fs::remove_file(partial).ok();
bail!("unexpected status {status} for {url} at offset {offset} — partial discarded");
}
_ => {
bail!("GET {url} returned {status}");
}
}
let written = resp
.copy_to(&mut file)
.with_context(|| format!("stream body of {url} (resumable — partial kept)"))?;
file.flush()?;
// fsync before the caller hashes/renames: the bytes must be durable.
file.sync_all()
.with_context(|| format!("fsync {}", partial.display()))?;
Ok(written)
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Read;
use std::net::TcpListener;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
/// A loopback HTTP/1.1 server for range semantics: serves `body`, honours
/// `Range: bytes=N-` with 206 unless `ignore_range`, and drops the
/// connection after `truncate_first` bytes on the first request (to force
/// a resume). No external crates, no real network.
struct RangeServer {
url: String,
hits: Arc<AtomicUsize>,
stop: Arc<AtomicBool>,
_job: gatling::background::Job<()>,
}
impl RangeServer {
fn start(body: Vec<u8>, ignore_range: bool, truncate_first: Option<usize>) -> Self {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let url = format!("http://{}/artifact.bin", listener.local_addr().unwrap());
let hits = Arc::new(AtomicUsize::new(0));
let stop = Arc::new(AtomicBool::new(false));
let (h, s) = (hits.clone(), stop.clone());
// ROOT-LAW-#0: background IO thread through gatling's ONE spawn.
let job = gatling::background::Job::spawn(move || {
while !s.load(Ordering::Relaxed) {
let Ok((mut sock, _)) = listener.accept() else {
std::thread::sleep(Duration::from_millis(5));
continue;
};
sock.set_read_timeout(Some(Duration::from_secs(2))).ok();
let mut buf = Vec::new();
let mut tmp = [0u8; 1024];
while let Ok(n) = sock.read(&mut tmp) {
if n == 0 {
break;
}
buf.extend_from_slice(&tmp[..n]);
if buf.windows(4).any(|w| w == b"\r\n\r\n") {
break;
}
}
let text = String::from_utf8_lossy(&buf);
let range_from: Option<usize> = text
.lines()
.find_map(|l| {
l.to_lowercase()
.strip_prefix("range: bytes=")
.map(String::from)
})
.and_then(|r| r.split('-').next().and_then(|n| n.parse().ok()));
let hit = h.fetch_add(1, Ordering::SeqCst);
let (status, slice_from) = match range_from {
Some(n) if !ignore_range => ("206 Partial Content", n),
_ => ("200 OK", 0),
};
let payload = &body[slice_from.min(body.len())..];
// Content-Length always states the FULL remainder; a
// truncated write below is a mid-body connection drop.
let head = format!(
"HTTP/1.1 {status}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
payload.len()
);
sock.write_all(head.as_bytes()).ok();
let cut = match truncate_first {
Some(c) if hit == 0 => c.min(payload.len()),
_ => payload.len(),
};
sock.write_all(&payload[..cut]).ok();
sock.flush().ok();
}
});
RangeServer {
url,
hits,
stop,
_job: job,
}
}
}
impl Drop for RangeServer {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
}
}
fn quick() -> DownloadPolicy {
DownloadPolicy {
attempts: 3,
retry_pause: Duration::from_millis(10),
connect_timeout: Duration::from_secs(5),
}
}
fn body() -> Vec<u8> {
(0u32..4096).flat_map(|i| i.to_le_bytes()).collect()
}
#[test]
fn a_clean_transfer_lands_verified_and_atomic() {
let b = body();
let want = crate::archive::sha256_hex(&b);
let srv = RangeServer::start(b.clone(), false, None);
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("artifact.bin");
let got = download_resumable(&srv.url, &dest, Some(&want), &quick()).unwrap();
assert_eq!(got.sha256, want);
assert_eq!(got.bytes, b.len() as u64);
assert!(!got.resumed);
assert_eq!(got.attempts, 1);
assert!(!partial_path(&dest).exists(), "no partial left behind");
assert_eq!(fs::read(&dest).unwrap(), b);
}
#[test]
fn an_interrupted_transfer_resumes_with_range_and_hashes_clean() {
let b = body();
let want = crate::archive::sha256_hex(&b);
// First request is cut mid-body → the retry must send Range and get 206.
let srv = RangeServer::start(b.clone(), false, Some(1000));
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("artifact.bin");
let got = download_resumable(&srv.url, &dest, Some(&want), &quick()).unwrap();
assert_eq!(got.sha256, want, "reassembled bytes hash correctly");
assert!(got.resumed, "the second attempt continued the partial");
assert!(got.attempts >= 2);
assert_eq!(fs::read(&dest).unwrap(), b);
}
#[test]
fn a_server_that_ignores_range_restarts_rather_than_corrupting() {
let b = body();
let want = crate::archive::sha256_hex(&b);
// Cut the first transfer AND ignore ranges: the 200-after-Range must
// truncate the partial, not append a full body onto the prefix.
let srv = RangeServer::start(b.clone(), true, Some(1000));
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("artifact.bin");
let got = download_resumable(&srv.url, &dest, Some(&want), &quick()).unwrap();
assert_eq!(got.sha256, want, "restart produced exact bytes, no splice");
assert_eq!(fs::read(&dest).unwrap(), b);
}
#[test]
fn a_digest_mismatch_refuses_and_discards_the_partial() {
let b = body();
let srv = RangeServer::start(b, false, None);
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("artifact.bin");
let err =
download_resumable(&srv.url, &dest, Some(&"ab".repeat(32)), &quick()).unwrap_err();
assert!(err.to_string().contains("failed after"), "{err:#}");
assert!(!dest.exists(), "a wrong file never wears the final name");
assert!(
!partial_path(&dest).exists(),
"poisoned bytes are not kept for resume"
);
// Every attempt re-fetched (no resume onto poison).
assert_eq!(srv.hits.load(Ordering::SeqCst), 3);
}
#[test]
fn an_already_landed_file_costs_no_request() {
let b = body();
let want = crate::archive::sha256_hex(&b);
let srv = RangeServer::start(b.clone(), false, None);
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("artifact.bin");
fs::write(&dest, &b).unwrap();
let got = download_resumable(&srv.url, &dest, Some(&want), &quick()).unwrap();
assert_eq!(got.attempts, 0, "idempotent: no network for a landed file");
assert_eq!(srv.hits.load(Ordering::SeqCst), 0);
}
#[test]
fn a_stale_file_at_dest_is_replaced_not_trusted() {
let b = body();
let want = crate::archive::sha256_hex(&b);
let srv = RangeServer::start(b.clone(), false, None);
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("artifact.bin");
fs::write(&dest, b"old wrong bytes").unwrap();
let got = download_resumable(&srv.url, &dest, Some(&want), &quick()).unwrap();
assert_eq!(got.sha256, want);
assert_eq!(fs::read(&dest).unwrap(), b);
}
}