net: retry flaky archive fetches and pin index downloads via by-hash
Busy mirrors and CDNs routinely break bulk fetches: pooled keep-alive
connections get closed remotely ('error sending request'), downloads are
cut short (surfacing as bogus checksum mismatches), and index generations
momentarily drift from the Release file fetched moments before.
- shared client: short idle-pool timeout and TCP keepalive, and a
bounded-retry GET helper now used for index, Release, keyring and
Launchpad fetches (previously reqwest::get, which has no timeouts)
- downloads: retry the whole download, and check the content length so
truncation is reported as such instead of a checksum mismatch
- sources index: on a checksum mismatch against the Release file, retry
pinned to the exact listed generation via Debian's by-hash mechanism;
body-read errors are retried and reported per component instead of
aborting the whole lookup
This commit is contained in:
+4
-4
@@ -137,7 +137,7 @@ impl ChecksumKind {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Human-readable algorithm name, for error messages
|
/// Human-readable algorithm name, for error messages
|
||||||
fn name(self) -> &'static str {
|
pub fn name(self) -> &'static str {
|
||||||
match self {
|
match self {
|
||||||
ChecksumKind::Md5 => "MD5",
|
ChecksumKind::Md5 => "MD5",
|
||||||
ChecksumKind::Sha1 => "SHA-1",
|
ChecksumKind::Sha1 => "SHA-1",
|
||||||
@@ -782,7 +782,7 @@ pub async fn verify_suite(
|
|||||||
/// GET a URL, returning None on any HTTP error, non-success status, or body
|
/// GET a URL, returning None on any HTTP error, non-success status, or body
|
||||||
/// read failure (Release files are probed, so absence is a normal outcome)
|
/// read failure (Release files are probed, so absence is a normal outcome)
|
||||||
async fn fetch_optional(url: &str) -> Option<Vec<u8>> {
|
async fn fetch_optional(url: &str) -> Option<Vec<u8>> {
|
||||||
match reqwest::get(url).await {
|
match crate::distro_info::http_get_retried(url).await {
|
||||||
Ok(response) if response.status().is_success() => match response.bytes().await {
|
Ok(response) if response.status().is_success() => match response.bytes().await {
|
||||||
Ok(bytes) => return Some(bytes.to_vec()),
|
Ok(bytes) => return Some(bytes.to_vec()),
|
||||||
Err(e) => debug!("Reading the body of '{url}' failed: {e}"),
|
Err(e) => debug!("Reading the body of '{url}' failed: {e}"),
|
||||||
@@ -811,7 +811,7 @@ async fn fetch_keyring_cached(url: &str) -> Result<Vec<u8>, Box<dyn Error + Send
|
|||||||
return Ok(cached.clone());
|
return Ok(cached.clone());
|
||||||
}
|
}
|
||||||
|
|
||||||
let response = reqwest::get(url).await?;
|
let response = crate::distro_info::http_get_retried(url).await?;
|
||||||
if !response.status().is_success() {
|
if !response.status().is_success() {
|
||||||
return Err(format!(
|
return Err(format!(
|
||||||
"downloading keyring from '{url}' failed with HTTP {}",
|
"downloading keyring from '{url}' failed with HTTP {}",
|
||||||
@@ -887,7 +887,7 @@ pub async fn ppa_keyring_bytes(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let api_url = format!("https://api.launchpad.net/1.0/~{owner}/+archive/ubuntu/{name}");
|
let api_url = format!("https://api.launchpad.net/1.0/~{owner}/+archive/ubuntu/{name}");
|
||||||
let response = reqwest::get(&api_url).await?;
|
let response = crate::distro_info::http_get_retried(&api_url).await?;
|
||||||
if !response.status().is_success() {
|
if !response.status().is_success() {
|
||||||
return Err(format!(
|
return Err(format!(
|
||||||
"querying the Launchpad API for the signing key of PPA \
|
"querying the Launchpad API for the signing key of PPA \
|
||||||
|
|||||||
@@ -53,9 +53,14 @@ lazy_static! {
|
|||||||
|
|
||||||
// Shared HTTP client used for all outgoing plain requests: timeouts keep
|
// Shared HTTP client used for all outgoing plain requests: timeouts keep
|
||||||
// a hanging remote (connect or transfer) from stalling pkh indefinitely.
|
// a hanging remote (connect or transfer) from stalling pkh indefinitely.
|
||||||
|
// The short pool idle timeout and TCP keepalive avoid reusing keep-alive
|
||||||
|
// connections that the remote closed in the meantime, which surfaces as
|
||||||
|
// spurious 'error sending request' failures on busy mirrors/CDNs.
|
||||||
static ref HTTP_CLIENT: reqwest::Client = reqwest::Client::builder()
|
static ref HTTP_CLIENT: reqwest::Client = reqwest::Client::builder()
|
||||||
.connect_timeout(Duration::from_secs(10))
|
.connect_timeout(Duration::from_secs(10))
|
||||||
.timeout(Duration::from_secs(30))
|
.timeout(Duration::from_secs(30))
|
||||||
|
.pool_idle_timeout(Duration::from_secs(10))
|
||||||
|
.tcp_keepalive(Duration::from_secs(30))
|
||||||
.build()
|
.build()
|
||||||
.expect("building the shared HTTP client with static options cannot fail");
|
.expect("building the shared HTTP client with static options cannot fail");
|
||||||
}
|
}
|
||||||
@@ -66,6 +71,49 @@ pub(crate) fn http_client() -> &'static reqwest::Client {
|
|||||||
&HTTP_CLIENT
|
&HTTP_CLIENT
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// GET `url` with bounded retries on transient transport errors (a pooled
|
||||||
|
/// keep-alive connection closed by the remote, a momentary network hiccup,
|
||||||
|
/// ...): these always succeed again on a fresh connection, and mirrors are
|
||||||
|
/// busy enough that unguarded single attempts make bulk operations flaky.
|
||||||
|
///
|
||||||
|
/// The response status is not inspected: 404s and the like are meaningful
|
||||||
|
/// answers, not transport failures.
|
||||||
|
pub(crate) async fn http_get_retried(url: &str) -> reqwest::Result<reqwest::Response> {
|
||||||
|
http_get_retried_with_timeout(url, None).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// [`http_get_retried`] with a per-request timeout override, for large
|
||||||
|
/// streaming downloads that exceed the shared client's total timeout
|
||||||
|
pub(crate) async fn http_get_retried_with_timeout(
|
||||||
|
url: &str,
|
||||||
|
timeout: Option<Duration>,
|
||||||
|
) -> reqwest::Result<reqwest::Response> {
|
||||||
|
const ATTEMPTS: u32 = 3;
|
||||||
|
let mut last_error: Option<reqwest::Error> = None;
|
||||||
|
for attempt in 0..ATTEMPTS {
|
||||||
|
let mut request = http_client().get(url);
|
||||||
|
if let Some(timeout) = timeout {
|
||||||
|
request = request.timeout(timeout);
|
||||||
|
}
|
||||||
|
match request.send().await {
|
||||||
|
Ok(response) => return Ok(response),
|
||||||
|
Err(e) => {
|
||||||
|
if attempt + 1 < ATTEMPTS {
|
||||||
|
log::debug!(
|
||||||
|
"GET '{url}' failed (attempt {}/{}, retrying): {}",
|
||||||
|
attempt + 1,
|
||||||
|
ATTEMPTS,
|
||||||
|
e
|
||||||
|
);
|
||||||
|
tokio::time::sleep(Duration::from_millis(300 * (u64::from(attempt) + 1))).await;
|
||||||
|
}
|
||||||
|
last_error = Some(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(last_error.expect("at least one attempt was made"))
|
||||||
|
}
|
||||||
|
|
||||||
/// Parse an optional '%Y-%m-%d' date from a CSV cell, warning instead of
|
/// Parse an optional '%Y-%m-%d' date from a CSV cell, warning instead of
|
||||||
/// panicking on invalid remote data
|
/// panicking on invalid remote data
|
||||||
fn parse_optional_date(value: Option<&str>, series: &str, field: &str) -> Option<NaiveDate> {
|
fn parse_optional_date(value: Option<&str>, series: &str, field: &str) -> Option<NaiveDate> {
|
||||||
|
|||||||
+105
-18
@@ -387,30 +387,41 @@ async fn get(
|
|||||||
|
|
||||||
debug!("Fetching sources from: {}", url);
|
debug!("Fetching sources from: {}", url);
|
||||||
|
|
||||||
let response = match crate::distro_info::http_client().get(&url).send().await {
|
let compressed_data = match fetch_index_bytes(&url).await {
|
||||||
Ok(resp) => resp,
|
Ok(data) => data,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
debug!("Failed to fetch {}: {}", url, e);
|
debug!("Failed to fetch {url}: {e}");
|
||||||
fetch_errors.push(format!("{suite}/{component}: {}", e));
|
fetch_errors.push(format!("{suite}/{component}: {e}"));
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
if !response.status().is_success() {
|
|
||||||
debug!("Failed to fetch {}: status {}", url, response.status());
|
|
||||||
fetch_errors.push(format!("{suite}/{component}: HTTP {}", response.status()));
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
let compressed_data = response.bytes().await?;
|
|
||||||
|
|
||||||
// The index must match the checksums listed in the signed Release
|
// The index must match the checksums listed in the signed Release
|
||||||
// file: this is what closes the 'substituted index with matching
|
// file: this is what closes the 'substituted index with matching
|
||||||
// artifact checksums' man-in-the-middle attack
|
// artifact checksums' man-in-the-middle attack
|
||||||
let suite_rel_path = format!("{component}/source/Sources.gz");
|
let suite_rel_path = format!("{component}/source/Sources.gz");
|
||||||
verified
|
let compressed_data = match verified.verify_file(&suite_rel_path, &compressed_data) {
|
||||||
.verify_file(&suite_rel_path, &compressed_data)
|
Ok(()) => compressed_data,
|
||||||
.map_err(release::VerifyError)?;
|
Err(verify_error) => {
|
||||||
|
// Busy mirrors and CDNs can serve an index generation
|
||||||
|
// slightly older or newer than the Release file fetched
|
||||||
|
// moments before. Pin the exact generation listed in the
|
||||||
|
// Release file via Debian's by-hash mechanism before
|
||||||
|
// failing.
|
||||||
|
match fetch_index_by_hash(&url, &verified, &suite_rel_path).await {
|
||||||
|
Ok(pinned) => {
|
||||||
|
debug!(
|
||||||
|
"index at '{url}' did not match the Release file: used the by-hash copy"
|
||||||
|
);
|
||||||
|
pinned
|
||||||
|
}
|
||||||
|
Err(by_hash_error) => {
|
||||||
|
debug!("by-hash fetch of '{}' failed: {}", url, by_hash_error);
|
||||||
|
return Err(release::VerifyError(verify_error).into());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
debug!(
|
debug!(
|
||||||
"Downloaded Sources.gz for {}/{}/{}",
|
"Downloaded Sources.gz for {}/{}/{}",
|
||||||
@@ -577,10 +588,86 @@ async fn get_flat_repo_series(base_url: &str) -> Result<String, Box<dyn Error>>
|
|||||||
Err(format!("No Codename or Suite field in Release file at '{url}'").into())
|
Err(format!("No Codename or Suite field in Release file at '{url}'").into())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Fetch an index pinned to the exact generation listed in the Release file
|
||||||
|
///
|
||||||
|
/// Uses Debian's by-hash mechanism, supported by the main archives (Debian,
|
||||||
|
/// Ubuntu, PPAs): the server serves the index generation matching the
|
||||||
|
/// Release file instead of whatever the mirror/CDN currently holds. The
|
||||||
|
/// returned content is verified again, so a by-hash-incapable mirror can
|
||||||
|
/// only cause a 404 here, never a silent mismatch.
|
||||||
|
async fn fetch_index_by_hash(
|
||||||
|
index_url: &str,
|
||||||
|
verified: &VerifiedRelease,
|
||||||
|
rel_path: &str,
|
||||||
|
) -> Result<Vec<u8>, Box<dyn Error>> {
|
||||||
|
let entry = verified.hash_for(rel_path).ok_or_else(|| {
|
||||||
|
format!("'{rel_path}' is not listed in the Release file: cannot fetch it by hash")
|
||||||
|
})?;
|
||||||
|
// By-hash URLs use the Release field names (MD5Sum, SHA1, SHA256, ...)
|
||||||
|
let algo = match entry.kind {
|
||||||
|
release::ChecksumKind::Md5 => "MD5Sum",
|
||||||
|
release::ChecksumKind::Sha1 => "SHA1",
|
||||||
|
release::ChecksumKind::Sha256 => "SHA256",
|
||||||
|
release::ChecksumKind::Sha512 => "SHA512",
|
||||||
|
};
|
||||||
|
// The by-hash layout replaces the file name with the hash reference:
|
||||||
|
// 'main/source/Sources.gz' is served at 'main/source/by-hash/SHA256/<digest>'
|
||||||
|
let (dir, _) = index_url
|
||||||
|
.rsplit_once('/')
|
||||||
|
.ok_or_else(|| format!("'{index_url}' has no parent directory"))?;
|
||||||
|
let url = format!("{dir}/by-hash/{algo}/{}", entry.hash);
|
||||||
|
|
||||||
|
let response = crate::distro_info::http_get_retried(&url).await?;
|
||||||
|
if !response.status().is_success() {
|
||||||
|
return Err(format!("HTTP {} for '{url}'", response.status()).into());
|
||||||
|
}
|
||||||
|
let data = response.bytes().await?.to_vec();
|
||||||
|
verified.verify_file(rel_path, &data)?;
|
||||||
|
Ok(data)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Fetch the compressed sources index at `url`
|
||||||
|
///
|
||||||
|
/// Transient transport failures and truncated bodies are retried; a
|
||||||
|
/// non-success HTTP status is a meaningful answer (e.g. a component that
|
||||||
|
/// does not exist in the suite) and is reported without retrying.
|
||||||
|
async fn fetch_index_bytes(url: &str) -> Result<Vec<u8>, String> {
|
||||||
|
const ATTEMPTS: u32 = 3;
|
||||||
|
let mut last_error = String::new();
|
||||||
|
for attempt in 1..=ATTEMPTS {
|
||||||
|
match crate::distro_info::http_get_retried(url).await {
|
||||||
|
Err(e) => {
|
||||||
|
last_error = e.to_string();
|
||||||
|
log::debug!("fetch of '{url}' failed (attempt {attempt}/{ATTEMPTS}): {last_error}");
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(300 * u64::from(attempt))).await;
|
||||||
|
}
|
||||||
|
Ok(response) => {
|
||||||
|
if !response.status().is_success() {
|
||||||
|
return Err(format!("HTTP {}", response.status()));
|
||||||
|
}
|
||||||
|
match response.bytes().await {
|
||||||
|
Ok(bytes) => return Ok(bytes.to_vec()),
|
||||||
|
Err(e) => {
|
||||||
|
// Truncated or corrupted body: retry from scratch
|
||||||
|
last_error = e.to_string();
|
||||||
|
log::debug!(
|
||||||
|
"reading the body of '{url}' failed (attempt {attempt}/{ATTEMPTS}): {last_error}"
|
||||||
|
);
|
||||||
|
tokio::time::sleep(
|
||||||
|
std::time::Duration::from_millis(300 * u64::from(attempt)),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(last_error)
|
||||||
|
}
|
||||||
|
|
||||||
/// Fetch the sources index of a flat repository
|
/// Fetch the sources index of a flat repository
|
||||||
///
|
///
|
||||||
/// Flat repositories are free to serve any compressed variant of the index
|
/// Flat repositories are free to serve any compressed variant of the index/// (or an uncompressed one), so try the usual candidates in turn. When the
|
||||||
/// (or an uncompressed one), so try the usual candidates in turn. When the
|
|
||||||
/// repository published a Release file, each index candidate is
|
/// repository published a Release file, each index candidate is
|
||||||
/// checksum-verified against it (failing on mismatch, since the artifact
|
/// checksum-verified against it (failing on mismatch, since the artifact
|
||||||
/// hashes would come from the index itself).
|
/// hashes would come from the index itself).
|
||||||
@@ -592,7 +679,7 @@ async fn get_flat_repo_sources(
|
|||||||
let mut errors = Vec::new();
|
let mut errors = Vec::new();
|
||||||
for name in ["Sources.xz", "Sources.gz", "Sources"] {
|
for name in ["Sources.xz", "Sources.gz", "Sources"] {
|
||||||
let url = format!("{base}/{name}");
|
let url = format!("{base}/{name}");
|
||||||
match crate::distro_info::http_client().get(&url).send().await {
|
match crate::distro_info::http_get_retried(&url).await {
|
||||||
Ok(response) if response.status().is_success() => {
|
Ok(response) if response.status().is_success() => {
|
||||||
let data = response.bytes().await?.to_vec();
|
let data = response.bytes().await?.to_vec();
|
||||||
|
|
||||||
|
|||||||
+56
-14
@@ -309,23 +309,53 @@ async fn download_file_checksum(
|
|||||||
algo: crate::package_info::ChecksumAlgo,
|
algo: crate::package_info::ChecksumAlgo,
|
||||||
target_dir: &Path,
|
target_dir: &Path,
|
||||||
progress: ProgressCallback<'_>,
|
progress: ProgressCallback<'_>,
|
||||||
|
) -> Result<(), Box<dyn Error>> {
|
||||||
|
// Archive mirrors and CDNs are busy enough that single attempts fail
|
||||||
|
// spuriously (dropped connections, truncated bodies, index generations
|
||||||
|
// momentarily out of sync): retry the whole download a few times before
|
||||||
|
// reporting the last failure.
|
||||||
|
const ATTEMPTS: u32 = 3;
|
||||||
|
let mut last_error: Box<dyn Error> = String::new().into();
|
||||||
|
for attempt in 1..=ATTEMPTS {
|
||||||
|
match download_file_checksum_once(url, checksum, algo, target_dir, progress).await {
|
||||||
|
Ok(()) => return Ok(()),
|
||||||
|
Err(e) => {
|
||||||
|
log::warn!(
|
||||||
|
"download of '{url}' failed (attempt {attempt}/{ATTEMPTS}): {e}"
|
||||||
|
);
|
||||||
|
last_error = e;
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(500 * u64::from(attempt))).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(format!(
|
||||||
|
"downloading '{url}' failed after {ATTEMPTS} attempts: {last_error}"
|
||||||
|
)
|
||||||
|
.into())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One download attempt of [`download_file_checksum`], verifying the
|
||||||
|
/// content length and the expected checksum
|
||||||
|
async fn download_file_checksum_once(
|
||||||
|
url: &str,
|
||||||
|
checksum: &str,
|
||||||
|
algo: crate::package_info::ChecksumAlgo,
|
||||||
|
target_dir: &Path,
|
||||||
|
progress: ProgressCallback<'_>,
|
||||||
) -> Result<(), Box<dyn Error>> {
|
) -> Result<(), Box<dyn Error>> {
|
||||||
// Download with the shared client (connect timeout). Large orig tarballs
|
// Download with the shared client (connect timeout). Large orig tarballs
|
||||||
// can legitimately take longer than the client's default total timeout,
|
// can legitimately take longer than the client's default total timeout,
|
||||||
// so use a generous per-request timeout for streaming downloads
|
// so use a generous per-request timeout for streaming downloads
|
||||||
let response = crate::distro_info::http_client()
|
let response = crate::distro_info::http_get_retried_with_timeout(
|
||||||
.get(url)
|
url,
|
||||||
.timeout(std::time::Duration::from_secs(30 * 60))
|
Some(std::time::Duration::from_secs(30 * 60)),
|
||||||
.send()
|
)
|
||||||
.await?;
|
.await
|
||||||
|
.map_err(|e| Box::new(e) as Box<dyn Error>)?;
|
||||||
if !response.status().is_success() {
|
if !response.status().is_success() {
|
||||||
return Err(format!("Failed to download '{}' : {}", url, response.status()).into());
|
return Err(format!("Failed to download '{}' : {}", url, response.status()).into());
|
||||||
}
|
}
|
||||||
|
let total_size = response.content_length();
|
||||||
let total_size = response
|
|
||||||
.content_length()
|
|
||||||
.ok_or(format!("Failed to get content length from '{}'", url))?;
|
|
||||||
let mut index = 0;
|
|
||||||
|
|
||||||
// Target file: extract file name from URL
|
// Target file: extract file name from URL
|
||||||
let filename = Path::new(url)
|
let filename = Path::new(url)
|
||||||
@@ -341,18 +371,30 @@ async fn download_file_checksum(
|
|||||||
let mut stream = response.bytes_stream();
|
let mut stream = response.bytes_stream();
|
||||||
// Accumulate the downloaded bytes so we can compute the final digest with the
|
// Accumulate the downloaded bytes so we can compute the final digest with the
|
||||||
// correct algorithm once the download is complete.
|
// correct algorithm once the download is complete.
|
||||||
let mut buffer: Vec<u8> = Vec::with_capacity(total_size as usize);
|
let mut buffer: Vec<u8> = Vec::with_capacity(total_size.unwrap_or(0) as usize);
|
||||||
while let Some(item) = stream.next().await {
|
while let Some(item) = stream.next().await {
|
||||||
let chunk = item?;
|
let chunk = item?;
|
||||||
file.write_all(&chunk)?;
|
file.write_all(&chunk)?;
|
||||||
buffer.extend_from_slice(&chunk);
|
buffer.extend_from_slice(&chunk);
|
||||||
|
|
||||||
if let Some(cb) = progress {
|
if let (Some(cb), Some(total)) = (progress, total_size) {
|
||||||
index = min(index + chunk.len(), total_size as usize);
|
let index = min(buffer.len(), total as usize);
|
||||||
cb("", "Downloading...", index, total_size as usize);
|
cb("", "Downloading...", index, total as usize);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A dropped connection can end the stream early: never hand a truncated
|
||||||
|
// file to the checksum check (its mismatch message would hide the cause)
|
||||||
|
if let Some(total) = total_size
|
||||||
|
&& buffer.len() != total as usize
|
||||||
|
{
|
||||||
|
return Err(format!(
|
||||||
|
"incomplete download from '{url}': got {} of {total} bytes",
|
||||||
|
buffer.len()
|
||||||
|
)
|
||||||
|
.into());
|
||||||
|
}
|
||||||
|
|
||||||
// Verify checksum using the algorithm specified for this file
|
// Verify checksum using the algorithm specified for this file
|
||||||
let calculated_checksum = algo.hex_digest(&buffer);
|
let calculated_checksum = algo.hex_digest(&buffer);
|
||||||
if calculated_checksum != checksum {
|
if calculated_checksum != checksum {
|
||||||
|
|||||||
Reference in New Issue
Block a user