diff --git a/packages/oneclient_core/tests/download_verify.rs b/packages/oneclient_core/tests/download_verify.rs index 7c741a46..c859e44e 100644 --- a/packages/oneclient_core/tests/download_verify.rs +++ b/packages/oneclient_core/tests/download_verify.rs @@ -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, 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, + ok_body: Vec, +) -> (String, std::sync::Arc) { + 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"NoSuchKey".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() { diff --git a/packages/oneclient_net/src/download.rs b/packages/oneclient_net/src/download.rs index a59a2609..eda8e99c 100644 --- a/packages/oneclient_net/src/download.rs +++ b/packages/oneclient_net/src/download.rs @@ -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(()) diff --git a/packages/oneclient_net/src/error.rs b/packages/oneclient_net/src/error.rs index d64cf0f6..653fc584 100644 --- a/packages/oneclient_net/src/error.rs +++ b/packages/oneclient_net/src/error.rs @@ -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), diff --git a/packages/oneclient_net/src/service.rs b/packages/oneclient_net/src/service.rs index eb4cbb81..30aeb9f5 100644 --- a/packages/oneclient_net/src/service.rs +++ b/packages/oneclient_net/src/service.rs @@ -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);