fix: downloading errors should be properly handled now
This commit is contained in:
parent
e272c66696
commit
885bd31f44
4 changed files with 299 additions and 13 deletions
|
|
@ -49,6 +49,243 @@ async fn sha1_file_handles_empty_file() {
|
|||
assert_eq!(sha1_file(&path).await.unwrap(), sha1_bytes(&[]));
|
||||
}
|
||||
|
||||
/// Serves `body` with a 200, declaring `content_length` however long `body` is —
|
||||
/// pass a larger one to imitate a mirror holding the wrong file.
|
||||
async fn serve(body: Vec<u8>, content_length: usize) -> String {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind");
|
||||
let port = listener.local_addr().expect("addr").port();
|
||||
|
||||
tokio::spawn(async move {
|
||||
while let Ok((mut stream, _)) = listener.accept().await {
|
||||
let body = body.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut request = [0u8; 1024];
|
||||
let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut request).await;
|
||||
|
||||
let head = format!("HTTP/1.1 200 OK\r\nContent-Length: {content_length}\r\n\r\n");
|
||||
let _ = stream.write_all(head.as_bytes()).await;
|
||||
let _ = stream.write_all(&body).await;
|
||||
let _ = stream.flush().await;
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
format!("http://127.0.0.1:{port}/artifact.jar")
|
||||
}
|
||||
|
||||
/// Answers the first `failures` requests with `status`, then serves `ok_body`.
|
||||
/// Returns the URL and a request counter. `usize::MAX` never recovers.
|
||||
async fn serve_flaky(
|
||||
status: u16,
|
||||
reason: &'static str,
|
||||
failures: usize,
|
||||
error_body: Vec<u8>,
|
||||
ok_body: Vec<u8>,
|
||||
) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind");
|
||||
let port = listener.local_addr().expect("addr").port();
|
||||
let served = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
tokio::spawn({
|
||||
let served = Arc::clone(&served);
|
||||
async move {
|
||||
while let Ok((mut stream, _)) = listener.accept().await {
|
||||
let (error_body, ok_body) = (error_body.clone(), ok_body.clone());
|
||||
let served = Arc::clone(&served);
|
||||
|
||||
tokio::spawn(async move {
|
||||
// Draining the request first, or the client sees a reset.
|
||||
let mut request = [0u8; 1024];
|
||||
let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut request).await;
|
||||
|
||||
let seen = served.fetch_add(1, Ordering::SeqCst);
|
||||
let (status, reason, body) = if seen < failures {
|
||||
(status, reason, error_body)
|
||||
} else {
|
||||
(200, "OK", ok_body)
|
||||
};
|
||||
|
||||
let head = format!(
|
||||
"HTTP/1.1 {status} {reason}\r\nContent-Length: {}\r\n\r\n",
|
||||
body.len()
|
||||
);
|
||||
let _ = stream.write_all(head.as_bytes()).await;
|
||||
let _ = stream.write_all(&body).await;
|
||||
let _ = stream.flush().await;
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
(format!("http://127.0.0.1:{port}/artifact.jar"), served)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn download_to_path_never_writes_an_error_response_to_disk() {
|
||||
let services = dev::ephemeral_services().await.unwrap();
|
||||
let progress = GroupedProgressSession::start(&services.events, "test");
|
||||
let dir = Scratch::new("download-404");
|
||||
let dest = dir.join("artifact.jar");
|
||||
|
||||
// What a CDN answers for a missing object: well-formed, but XML not a jar.
|
||||
let (url, served) = serve_flaky(
|
||||
404,
|
||||
"Not Found",
|
||||
usize::MAX,
|
||||
b"<Error><Code>NoSuchKey</Code></Error>".to_vec(),
|
||||
Vec::new(),
|
||||
)
|
||||
.await;
|
||||
|
||||
let err = download_to_path(
|
||||
&services.requester,
|
||||
&services.events,
|
||||
&progress,
|
||||
"artifact",
|
||||
TaskCategory::Libraries,
|
||||
0,
|
||||
&url,
|
||||
&dest,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect_err("a 404 must not be saved as the file");
|
||||
|
||||
assert!(err.to_string().contains("404"), "{err}");
|
||||
assert!(err.to_string().contains("NoSuchKey"), "{err}");
|
||||
assert!(!dest.exists(), "an error body must never reach disk");
|
||||
// A 404 answers the same way every time; retrying it only delays the error.
|
||||
assert_eq!(served.load(std::sync::atomic::Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn download_to_path_retries_a_server_error_and_recovers() {
|
||||
let services = dev::ephemeral_services().await.unwrap();
|
||||
let progress = GroupedProgressSession::start(&services.events, "test");
|
||||
let dir = Scratch::new("download-500");
|
||||
let dest = dir.join("artifact.jar");
|
||||
|
||||
// Unlike a 404, a 500 can succeed on the next attempt.
|
||||
let (url, served) = serve_flaky(
|
||||
500,
|
||||
"Internal Server Error",
|
||||
1,
|
||||
b"boom".to_vec(),
|
||||
vec![7u8; 128],
|
||||
)
|
||||
.await;
|
||||
|
||||
download_to_path(
|
||||
&services.requester,
|
||||
&services.events,
|
||||
&progress,
|
||||
"artifact",
|
||||
TaskCategory::Libraries,
|
||||
128,
|
||||
&url,
|
||||
&dest,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("a retried server error should recover");
|
||||
|
||||
assert_eq!(std::fs::metadata(&dest).unwrap().len(), 128);
|
||||
assert_eq!(served.load(std::sync::atomic::Ordering::SeqCst), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn download_to_path_rejects_a_body_shorter_than_the_manifest_size() {
|
||||
let services = dev::ephemeral_services().await.unwrap();
|
||||
let progress = GroupedProgressSession::start(&services.events, "test");
|
||||
let dir = Scratch::new("download-short");
|
||||
let dest = dir.join("artifact.jar");
|
||||
|
||||
// Valid HTTP, just not the file the manifest described.
|
||||
let url = serve(vec![7u8; 64], 64).await;
|
||||
|
||||
let err = download_to_path(
|
||||
&services.requester,
|
||||
&services.events,
|
||||
&progress,
|
||||
"artifact",
|
||||
TaskCategory::Libraries,
|
||||
4096,
|
||||
&url,
|
||||
&dest,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect_err("a short body should fail without a hash to check");
|
||||
|
||||
assert!(err.to_string().contains("ended after 64 of 4096"), "{err}");
|
||||
assert!(!dest.exists(), "truncated download should be removed");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn download_to_path_accepts_an_unhashed_body_of_the_promised_length() {
|
||||
let services = dev::ephemeral_services().await.unwrap();
|
||||
let progress = GroupedProgressSession::start(&services.events, "test");
|
||||
let dir = Scratch::new("download-exact");
|
||||
let dest = dir.join("artifact.jar");
|
||||
|
||||
let url = serve(vec![7u8; 4096], 4096).await;
|
||||
|
||||
download_to_path(
|
||||
&services.requester,
|
||||
&services.events,
|
||||
&progress,
|
||||
"artifact",
|
||||
TaskCategory::Libraries,
|
||||
4096,
|
||||
&url,
|
||||
&dest,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("a body of the promised length should be kept");
|
||||
|
||||
assert_eq!(std::fs::metadata(&dest).unwrap().len(), 4096);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn download_to_path_keeps_an_unhashed_body_of_unknown_length() {
|
||||
let services = dev::ephemeral_services().await.unwrap();
|
||||
let progress = GroupedProgressSession::start(&services.events, "test");
|
||||
let dir = Scratch::new("download-unknown");
|
||||
let dest = dir.join("artifact.jar");
|
||||
|
||||
let url = serve(vec![7u8; 128], 128).await;
|
||||
|
||||
// Nothing declared a size, so there is nothing to check against.
|
||||
download_to_path(
|
||||
&services.requester,
|
||||
&services.events,
|
||||
&progress,
|
||||
"artifact",
|
||||
TaskCategory::Libraries,
|
||||
0,
|
||||
&url,
|
||||
&dest,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("an unmeasurable download should still be written");
|
||||
|
||||
assert_eq!(std::fs::metadata(&dest).unwrap().len(), 128);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires network"]
|
||||
async fn download_to_path_accepts_matching_hash() {
|
||||
|
|
|
|||
|
|
@ -23,14 +23,16 @@ const MAX_DOWNLOAD_ATTEMPTS: u32 = 3;
|
|||
|
||||
/// Whether fetching the same URL again has a realistic chance of working.
|
||||
///
|
||||
/// A transport error or a hash mismatch means the bytes arrived wrong, which is
|
||||
/// exactly what a retry fixes. Everything else — a 404, a bad URL, a full disk —
|
||||
/// will fail the same way every time, and retrying only multiplies the delay
|
||||
/// before the user sees the real error.
|
||||
/// A transport error, a hash mismatch, or a short body means the bytes arrived
|
||||
/// wrong, which is exactly what a retry fixes. Everything else — a 404, a bad
|
||||
/// URL, a full disk — will fail the same way every time, and retrying only
|
||||
/// multiplies the delay before the user sees the real error.
|
||||
fn worth_retrying(err: &RequestError) -> bool {
|
||||
matches!(
|
||||
err,
|
||||
RequestError::ReqwestError(_) | RequestError::HashMismatch { .. }
|
||||
RequestError::ReqwestError(_)
|
||||
| RequestError::HashMismatch { .. }
|
||||
| RequestError::IncompleteBody { .. }
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -149,12 +151,24 @@ async fn download_attempt(
|
|||
let request = reqwest::Request::new(Method::GET, url.parse()?);
|
||||
let response = client.send(request).await?;
|
||||
|
||||
// Prefer the manifest size; fall back to Content-Length only when unknown.
|
||||
let total = if expected_size > 0 {
|
||||
expected_size
|
||||
// Without this, an error body is written to disk under the requested file's name.
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
let bytes = response.bytes().await?;
|
||||
return Err(RequestError::HttpStatus {
|
||||
status: status.as_u16(),
|
||||
url: url.to_string(),
|
||||
snippet: crate::error::body_snippet(&bytes),
|
||||
});
|
||||
}
|
||||
|
||||
// Prefer the manifest size; Content-Length is absent for compressed or chunked bodies.
|
||||
let declared = if expected_size > 0 {
|
||||
Some(expected_size)
|
||||
} else {
|
||||
response.content_length().unwrap_or(0).max(1)
|
||||
response.content_length()
|
||||
};
|
||||
let total = declared.unwrap_or(0).max(1);
|
||||
|
||||
let options = ResponseOptions {
|
||||
notify: notify.cloned(),
|
||||
|
|
@ -162,10 +176,14 @@ async fn download_attempt(
|
|||
let stream = response.stream(options, events).await?;
|
||||
|
||||
let mut hasher = expected.map(|expected| ChecksumStream::new(expected.algorithm));
|
||||
let mut received = 0u64;
|
||||
{
|
||||
let stream = futures_lite::StreamExt::map(stream, |item| {
|
||||
if let (Ok(chunk), Some(hasher)) = (&item, hasher.as_mut()) {
|
||||
hasher.update(chunk);
|
||||
if let Ok(chunk) = &item {
|
||||
received += chunk.len() as u64;
|
||||
if let Some(hasher) = hasher.as_mut() {
|
||||
hasher.update(chunk);
|
||||
}
|
||||
}
|
||||
item
|
||||
});
|
||||
|
|
@ -191,6 +209,16 @@ async fn download_attempt(
|
|||
actual,
|
||||
});
|
||||
}
|
||||
} else if let Some(declared) = declared
|
||||
&& received != declared
|
||||
{
|
||||
// With no hash, the byte count is the only evidence the transfer finished.
|
||||
let _ = polyio::remove_file(dest).await;
|
||||
return Err(RequestError::IncompleteBody {
|
||||
source_desc: dest.display().to_string(),
|
||||
expected: declared,
|
||||
actual: received,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(())
|
||||
|
|
|
|||
|
|
@ -61,6 +61,15 @@ pub enum RequestError {
|
|||
actual: String,
|
||||
},
|
||||
|
||||
/// A download ended early without the transport reporting an error. Only
|
||||
/// reachable when no hash was supplied; otherwise it fails as a mismatch.
|
||||
#[error("{source_desc} ended after {actual} of {expected} bytes")]
|
||||
IncompleteBody {
|
||||
source_desc: String,
|
||||
expected: u64,
|
||||
actual: u64,
|
||||
},
|
||||
|
||||
#[error("Invalid URL: {0}")]
|
||||
UrlParseError(#[from] url::ParseError),
|
||||
|
||||
|
|
|
|||
|
|
@ -186,8 +186,7 @@ impl RequestClient {
|
|||
match self.client.execute(active_request).await {
|
||||
Ok(response) => {
|
||||
let status = response.status();
|
||||
let retryable = status.as_u16() == 429
|
||||
|| matches!(status.as_u16(), 502..=504);
|
||||
let retryable = status.as_u16() == 429 || status.is_server_error();
|
||||
|
||||
if retryable && cloneable && throttle_retries < MAX_THROTTLE_RETRIES {
|
||||
throttle_retries += 1;
|
||||
|
|
@ -241,6 +240,19 @@ impl RequestClient {
|
|||
events: &EventBus,
|
||||
) -> Result<(), RequestError> {
|
||||
let res = self.send(request).await?;
|
||||
|
||||
// Without this, an error body is written to disk under the requested file's name.
|
||||
let status = res.status();
|
||||
if !status.is_success() {
|
||||
let url = res.url().to_string();
|
||||
let bytes = res.bytes().await?;
|
||||
return Err(RequestError::HttpStatus {
|
||||
status: status.as_u16(),
|
||||
url,
|
||||
snippet: body_snippet(&bytes),
|
||||
});
|
||||
}
|
||||
|
||||
let size_hint = res.content_length();
|
||||
let http_stream = res.stream(options, events).await?;
|
||||
let http_stream = std::pin::pin!(http_stream);
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue