diff --git a/src/package_info.rs b/src/package_info.rs index 5a743c3..368979c 100644 --- a/src/package_info.rs +++ b/src/package_info.rs @@ -417,7 +417,10 @@ async fn get( } Err(by_hash_error) => { debug!("by-hash fetch of '{}' failed: {}", url, by_hash_error); - return Err(release::VerifyError(verify_error).into()); + return Err(release::VerifyError(format!( + "{verify_error}; the by-hash retry also failed: {by_hash_error}" + )) + .into()); } } } @@ -622,6 +625,9 @@ async fn fetch_index_by_hash( return Err(format!("HTTP {} for '{url}'", response.status()).into()); } let data = response.bytes().await?.to_vec(); + if data.is_empty() { + return Err(format!("empty body for '{url}'").into()); + } verified.verify_file(rel_path, &data)?; Ok(data) } @@ -646,6 +652,18 @@ async fn fetch_index_bytes(url: &str) -> Result, String> { return Err(format!("HTTP {}", response.status())); } match response.bytes().await { + Ok(bytes) if bytes.is_empty() => { + // CDNs occasionally answer 200 with an empty body + // under load: never a valid index, retry from scratch + last_error = "server returned an empty body".to_string(); + log::debug!( + "empty body for '{url}' (attempt {attempt}/{ATTEMPTS}), retrying" + ); + tokio::time::sleep( + std::time::Duration::from_millis(300 * u64::from(attempt)), + ) + .await; + } Ok(bytes) => return Ok(bytes.to_vec()), Err(e) => { // Truncated or corrupted body: retry from scratch @@ -924,6 +942,58 @@ pub async fn lookup( mod tests { use super::*; + /// Serve canned byte responses on a local port, one per connection (the + /// last response repeats), and return the base URL + /// + /// The canned responses must use 'Connection: close' so the client opens + /// a fresh connection (and receives a fresh response) per request. + fn serve_responses(responses: Vec>) -> String { + use std::io::{Read, Write}; + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + let mut served = 0usize; + for stream in listener.incoming().flatten() { + let index = served.min(responses.len() - 1); + served += 1; + let mut stream = stream; + // Drain the request first: closing with unread inbound data + // would send a TCP RST and destroy the response in flight + let mut buf = [0u8; 4096]; + loop { + match stream.read(&mut buf) { + Ok(0) => break, + Ok(n) if buf[..n].windows(4).any(|w| w == b"\r\n\r\n") => break, + Ok(_) => continue, + Err(_) => break, + } + } + let _ = stream.write_all(&responses[index]); + let _ = stream.flush(); + } + }); + format!("http://{addr}/Sources.gz") + } + + /// A CDN answering 200 with an empty body (observed under load) must be + /// retried instead of failing the index checksum verification + #[tokio::test] + async fn fetch_index_bytes_retries_empty_body() { + let valid = b"Package: hello\nVersion: 1.0\n\n"; + let empty_response = + b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_vec(); + let mut valid_response = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + valid.len() + ) + .into_bytes(); + valid_response.extend_from_slice(valid); + + let url = serve_responses(vec![empty_response, valid_response]); + let data = fetch_index_bytes(&url).await.unwrap(); + assert_eq!(data, valid); + } + #[tokio::test] async fn test_check_launchpad_repo() { // "hello" should exist on Launchpad for Ubuntu