diff --git a/crates/uv-bin-install/src/lib.rs b/crates/uv-bin-install/src/lib.rs index ba5965a99..15f34e094 100644 --- a/crates/uv-bin-install/src/lib.rs +++ b/crates/uv-bin-install/src/lib.rs @@ -3,7 +3,9 @@ //! These utilities are specifically for consuming distributions that are _not_ Python packages, //! e.g., `ruff` (which does have a Python package, but also has standalone binaries on GitHub). +use std::error::Error as _; use std::fmt; +use std::io; use std::path::PathBuf; use std::pin::Pin; use std::str::FromStr; @@ -11,16 +13,18 @@ use std::task::{Context, Poll}; use std::time::{Duration, SystemTimeError}; use futures::{StreamExt, TryStreamExt}; +use reqwest_retry::Retryable; use reqwest_retry::policies::ExponentialBackoff; use serde::Deserialize; use thiserror::Error; use tokio::io::{AsyncRead, ReadBuf}; use tokio_util::compat::FuturesAsyncReadCompatExt; use url::Url; +use uv_client::retryable_on_request_failure; use uv_distribution_filename::SourceDistExtension; use uv_cache::{Cache, CacheBucket, CacheEntry, Error as CacheError}; -use uv_client::{BaseClient, RetryState}; +use uv_client::{BaseClient, RetriableError, RetryState, fetch_with_url_fallback}; use uv_extract::{Error as ExtractError, stream}; use uv_pep440::{Version, VersionSpecifier, VersionSpecifiers}; use uv_platform::Platform; @@ -58,20 +62,47 @@ impl Binary { } } - /// Get the download URL for a specific version and platform. - pub fn download_url( + /// Get the ordered list of download URLs for a specific version and platform. + /// + /// The default Astral mirror is returned first, followed by the canonical GitHub URL as a + /// fallback. + pub fn download_urls( &self, version: &Version, platform: &str, format: ArchiveFormat, - ) -> Result { + ) -> Result, Error> { match self { Self::Ruff => { - let url = format!( - "https://github.com/astral-sh/ruff/releases/download/{version}/ruff-{platform}.{}", - format.extension() - ); - Url::parse(&url).map_err(|err| Error::UrlParse { url, source: err }) + let suffix = format!("{version}/ruff-{platform}.{}", format.extension()); + let canonical = format!("{RUFF_GITHUB_URL_PREFIX}{suffix}"); + let mirror = format!("{RUFF_DEFAULT_MIRROR}{suffix}"); + Ok(vec![ + DisplaySafeUrl::parse(&mirror).map_err(|err| Error::UrlParse { + url: mirror, + source: err, + })?, + DisplaySafeUrl::parse(&canonical).map_err(|err| Error::UrlParse { + url: canonical, + source: err, + })?, + ]) + } + } + } + + /// Given a canonical artifact URL (e.g., from the versions manifest), return an ordered list + /// of URLs to try: the default Astral mirror first, then the canonical URL as a fallback. + fn mirror_urls(self, canonical_url: DisplaySafeUrl) -> Vec { + match self { + Self::Ruff => { + if let Some(suffix) = canonical_url.as_str().strip_prefix(RUFF_GITHUB_URL_PREFIX) { + let mirror_str = format!("{RUFF_DEFAULT_MIRROR}{suffix}"); + if let Ok(mirror_url) = DisplaySafeUrl::parse(&mirror_str) { + return vec![mirror_url, canonical_url]; + } + } + vec![canonical_url] } } } @@ -155,6 +186,15 @@ impl fmt::Display for BinVersion { } } +/// The canonical GitHub URL prefix for Ruff releases. +const RUFF_GITHUB_URL_PREFIX: &str = "https://github.com/astral-sh/ruff/releases/download/"; + +/// The default Astral mirror for Ruff releases. +/// +/// This mirror is tried first for Ruff downloads. If it fails, uv falls back to the canonical +/// GitHub URL. +const RUFF_DEFAULT_MIRROR: &str = "https://releases.astral.sh/github/ruff/releases/download/"; + /// Base URL for the versions manifest. const VERSIONS_MANIFEST_URL: &str = "https://raw.githubusercontent.com/astral-sh/versions/main/v1"; @@ -188,15 +228,17 @@ struct BinArtifact { pub struct ResolvedVersion { /// The version number. pub version: Version, - /// The download URL for this version and current platform. - pub artifact_url: Url, + /// The ordered list of download URLs to try for this version and current platform. + /// + /// The default Astral mirror is listed first, with the canonical GitHub URL as a fallback. + pub artifact_urls: Vec, /// The archive format. pub archive_format: ArchiveFormat, } impl ResolvedVersion { /// Construct a [`ResolvedVersion`] from a [`Binary`] and a [`Version`] by inferring the - /// download URL and archive format from the current platform. + /// download URLs and archive format from the current platform. pub fn from_version(binary: Binary, version: Version) -> Result { let platform = Platform::from_env()?; let platform_name = platform.as_cargo_dist_triple(); @@ -205,10 +247,10 @@ impl ResolvedVersion { } else { ArchiveFormat::TarGz }; - let artifact_url = binary.download_url(&version, &platform_name, archive_format)?; + let artifact_urls = binary.download_urls(&version, &platform_name, archive_format)?; Ok(Self { version, - artifact_url, + artifact_urls, archive_format, }) } @@ -224,11 +266,18 @@ pub enum Error { source: reqwest_middleware::Error, }, + #[error("Failed to read from: {url}")] + Stream { + url: DisplaySafeUrl, + #[source] + source: reqwest::Error, + }, + #[error("Failed to parse URL: {url}")] UrlParse { url: String, #[source] - source: url::ParseError, + source: uv_redacted::DisplaySafeUrlError, }, #[error("Failed to extract archive")] @@ -298,17 +347,53 @@ pub enum Error { SystemTime(#[from] SystemTimeError), } -impl Error { - /// Return the number of retries that were made to complete this request before this error was - /// returned. - /// - /// Note that e.g. 3 retries equates to 4 attempts. +impl RetriableError for Error { fn retries(&self) -> u32 { if let Self::RetriedError { retries, .. } = self { return *retries; } 0 } + + /// Returns `true` if trying an alternative URL makes sense after this error. + /// + /// All errors arising from downloading (including streaming during extraction) + /// qualify. + fn should_try_next_url(&self) -> bool { + match self { + Self::Download { .. } => true, + Self::Stream { .. } => true, + Self::RetriedError { err, .. } => err.should_try_next_url(), + err => { + // Walk the error chain to see if there's a nested download or streaming error. + let mut source = err.source(); + while let Some(err) = source { + if let Some(io_err) = err.downcast_ref::() { + if io_err + .get_ref() + .and_then(|e| e.downcast_ref::() as Option<&Self>) + .is_some_and(|e| { + matches!(e, Self::Stream { .. } | Self::Download { .. }) + }) + { + return true; + } + } + source = err.source(); + } + // Make sure all retriable errors also trigger a fallback to the next URL. + retryable_on_request_failure(err) == Some(Retryable::Transient) + } + } + } + + fn into_retried(self, retries: u32, duration: Duration) -> Self { + Self::RetriedError { + err: Box::new(self), + retries, + duration, + } + } } /// Find a version of a binary that matches the given constraints. @@ -331,10 +416,11 @@ pub async fn find_matching_version( let platform_name = platform.as_cargo_dist_triple(); let manifest_url = format!("{}/{}.ndjson", VERSIONS_MANIFEST_URL, binary.name()); - let manifest_url_parsed = Url::parse(&manifest_url).map_err(|source| Error::UrlParse { - url: manifest_url.clone(), - source, - })?; + let manifest_url_parsed = + DisplaySafeUrl::parse(&manifest_url).map_err(|source| Error::UrlParse { + url: manifest_url.clone(), + source, + })?; let mut retry_state = RetryState::start(*retry_policy, manifest_url_parsed.clone()); @@ -344,7 +430,6 @@ pub async fn find_matching_version( constraints, exclude_newer, &platform_name, - &manifest_url, &manifest_url_parsed, client, ) @@ -380,13 +465,12 @@ async fn fetch_and_find_matching_version( constraints: Option<&uv_pep440::VersionSpecifiers>, exclude_newer: Option, platform_name: &str, - manifest_url: &str, - manifest_url_parsed: &Url, + manifest_url: &DisplaySafeUrl, client: &BaseClient, ) -> Result { let response = client - .for_host(&DisplaySafeUrl::from_url(manifest_url_parsed.clone())) - .get(manifest_url_parsed.clone()) + .for_host(manifest_url) + .get(Url::from(manifest_url.clone())) .send() .await .map_err(|source| Error::ManifestFetch { @@ -409,6 +493,7 @@ async fn fetch_and_find_matching_version( } let version_info: BinVersionInfo = serde_json::from_str(line_str)?; Ok(check_version_match( + binary, &version_info, constraints, exclude_newer, @@ -463,6 +548,7 @@ async fn fetch_and_find_matching_version( /// Returns `Some(resolved)` if the version matches and an artifact is found, /// `None` if the version doesn't match or no artifact is available for the platform. fn check_version_match( + binary: Binary, version_info: &BinVersionInfo, constraints: Option<&uv_pep440::VersionSpecifiers>, exclude_newer: Option, @@ -489,7 +575,7 @@ fn check_version_match( continue; } - let Ok(artifact_url) = Url::parse(&artifact.url) else { + let Ok(canonical_url) = DisplaySafeUrl::parse(&artifact.url) else { continue; }; @@ -501,7 +587,7 @@ fn check_version_match( return Some(ResolvedVersion { version: version_info.version.clone(), - artifact_url, + artifact_urls: binary.mirror_urls(canonical_url), archive_format, }); } @@ -521,10 +607,10 @@ pub async fn bin_install( let platform = Platform::from_env()?; let platform_name = platform.as_cargo_dist_triple(); - bin_install_from_url( + bin_install_from_urls( binary, &resolved.version, - &resolved.artifact_url, + &resolved.artifact_urls, resolved.archive_format, &platform_name, client, @@ -535,11 +621,11 @@ pub async fn bin_install( .await } -/// Install a binary from a specific URL. -async fn bin_install_from_url( +/// Install a binary from an ordered list of URLs, trying each in sequence. +async fn bin_install_from_urls( binary: Binary, version: &Version, - download_url: &Url, + download_urls: &[DisplaySafeUrl], format: ArchiveFormat, platform_name: &str, client: &BaseClient, @@ -547,7 +633,6 @@ async fn bin_install_from_url( cache: &Cache, reporter: &dyn Reporter, ) -> Result { - let download_url = DisplaySafeUrl::from_url(download_url.clone()); let cache_entry = CacheEntry::new( cache .bucket(CacheBucket::Binaries) @@ -566,17 +651,23 @@ async fn bin_install_from_url( let cache_dir = cache_entry.dir(); fs_err::tokio::create_dir_all(&cache_dir).await?; - let path = download_and_unpack_with_retry( - binary, - version, - client, - retry_policy, - cache, - reporter, - platform_name, - format, - &download_url, - &cache_entry, + let path = fetch_with_url_fallback( + download_urls, + *retry_policy, + &format!("`{binary}`"), + |url| { + download_and_unpack( + binary, + version, + client, + cache, + reporter, + platform_name, + format, + url, + &cache_entry, + ) + }, ) .await?; @@ -598,59 +689,9 @@ async fn bin_install_from_url( Ok(path) } -/// Download and unpack a binary with retry on stream failures. -async fn download_and_unpack_with_retry( - binary: Binary, - version: &Version, - client: &BaseClient, - retry_policy: &ExponentialBackoff, - cache: &Cache, - reporter: &dyn Reporter, - platform_name: &str, - format: ArchiveFormat, - download_url: &DisplaySafeUrl, - cache_entry: &CacheEntry, -) -> Result { - let mut retry_state = RetryState::start(*retry_policy, download_url.clone()); - - loop { - let result = download_and_unpack( - binary, - version, - client, - cache, - reporter, - platform_name, - format, - download_url, - cache_entry, - ) - .await; - - match result { - Ok(path) => return Ok(path), - Err(err) => { - if let Some(backoff) = retry_state.should_retry(&err, err.retries()) { - retry_state.sleep_backoff(backoff).await; - continue; - } - return if retry_state.total_retries() > 0 { - Err(Error::RetriedError { - err: Box::new(err), - retries: retry_state.total_retries(), - duration: retry_state.duration()?, - }) - } else { - Err(err) - }; - } - } - } -} - -/// Download and unpackage a binary, +/// Download and unpack a binary from a single URL. /// -/// NOTE [`download_and_unpack_with_retry`] should be used instead. +/// Use [`bin_install_from_urls`] (via [`fetch_with_url_fallback`]) to get URL-fallback and retry. async fn download_and_unpack( binary: Binary, version: &Version, @@ -659,14 +700,14 @@ async fn download_and_unpack( reporter: &dyn Reporter, platform_name: &str, format: ArchiveFormat, - download_url: &DisplaySafeUrl, + download_url: DisplaySafeUrl, cache_entry: &CacheEntry, ) -> Result { // Create a temporary directory for extraction let temp_dir = tempfile::tempdir_in(cache.bucket(CacheBucket::Binaries))?; let response = client - .for_host(download_url) + .for_host(&download_url) .get(Url::from(download_url.clone())) .send() .await @@ -706,14 +747,19 @@ async fn download_and_unpack( // Stream download directly to extraction let reader = response .bytes_stream() - .map_err(std::io::Error::other) + .map_err(|err| { + std::io::Error::other(Error::Stream { + url: download_url.clone(), + source: err, + }) + }) .into_async_read() .compat(); let id = reporter.on_download_start(binary.name(), version, size); let mut progress_reader = ProgressReader::new(reader, id, reporter); stream::archive( - download_url, + &download_url, &mut progress_reader, format.into(), temp_dir.path(), @@ -794,3 +840,62 @@ where }) } } + +#[cfg(test)] +mod tests { + use std::io::Write; + use std::net::TcpListener; + use uv_client::{BaseClientBuilder, retryable_on_request_failure}; + use uv_redacted::DisplaySafeUrl; + + use super::*; + + /// Verify that `should_try_next_url` returns `true` even for streaming errors + /// that `retryable_on_request_failure` does not recognise as transient. + /// + /// This exercises a realistic body-streaming protocol failure: the server + /// advertises chunked transfer encoding but sends an invalid chunk size. + #[tokio::test] + async fn test_non_retryable_stream_error_triggers_url_fallback() { + use futures::TryStreamExt; + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + + std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut buf = [0u8; 4096]; + let _ = std::io::Read::read(&mut stream, &mut buf); + stream + .write_all( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\nZZZ\r\nhello\r\n0\r\n\r\n", + ) + .unwrap(); + }); + + let url = DisplaySafeUrl::parse(&format!("http://{addr}/ruff.tar.gz")).unwrap(); + let client = BaseClientBuilder::default().build(); + let response = client + .for_host(&url) + .get(Url::from(url.clone())) + .send() + .await + .unwrap(); + + let reqwest_err = response.bytes_stream().try_next().await.unwrap_err(); + assert!(reqwest_err.is_body() || reqwest_err.is_decode()); + + let err = Error::Extract { + source: ExtractError::Io(io::Error::other(Error::Stream { + url, + source: reqwest_err, + })), + }; + + assert!(retryable_on_request_failure(&err).is_none()); + assert!( + err.should_try_next_url(), + "non-retryable streaming error should still trigger URL fallback, got: {err}" + ); + } +} diff --git a/crates/uv-client/src/base_client.rs b/crates/uv-client/src/base_client.rs index 399525cbd..0fbfd1d25 100644 --- a/crates/uv-client/src/base_client.rs +++ b/crates/uv-client/src/base_client.rs @@ -23,8 +23,10 @@ use reqwest_retry::{ Jitter, RetryPolicy, RetryTransientMiddleware, Retryable, RetryableStrategy, default_on_request_error, default_on_request_success, }; + use thiserror::Error; -use tracing::{debug, trace}; + +use tracing::{debug, trace, warn}; use url::ParseError; use url::Url; @@ -1373,6 +1375,74 @@ impl RetryState { } } +/// An error type that supports URL-fallback and exponential-backoff retry logic. +/// +/// Used by [`fetch_with_url_fallback`] to drive the retry loop without knowing the concrete error +/// type. +pub trait RetriableError: std::error::Error + Sized + 'static { + /// Returns `true` if an alternative URL should be tried immediately (without backoff). + fn should_try_next_url(&self) -> bool; + + /// Returns the number of inner retries already recorded in this error. + fn retries(&self) -> u32; + + /// Wrap the error to indicate that the operation was retried `retries` times before failing. + #[must_use] + fn into_retried(self, retries: u32, duration: Duration) -> Self; +} + +/// Try a fallible async operation against each URL in order, with exponential backoff. +/// +/// URLs are tried in sequence without any backoff between them. Backoff is only applied after all +/// URLs have been exhausted. On the next retry attempt the full URL list is tried again from the +/// beginning. +pub async fn fetch_with_url_fallback( + urls: &[DisplaySafeUrl], + retry_policy: ExponentialBackoff, + subject: &str, + mut attempt: F, +) -> Result +where + F: AsyncFnMut(DisplaySafeUrl) -> Result, + E: RetriableError + From, +{ + let mut retry_state = RetryState::start( + retry_policy, + // The last URL will trigger backoff if it fails. + urls.last().expect("urls must not be empty").clone(), + ); + + 'retry: loop { + for (i, url) in urls.iter().enumerate() { + let is_last = i == urls.len() - 1; + match attempt(url.clone()).await { + Ok(result) => return Ok(result), + Err(err) => { + if !is_last && err.should_try_next_url() { + warn!( + "Failed to fetch {subject} from {url} ({err}); falling back to {}", + urls[i + 1] + ); + continue; + } + // All URLs exhausted; apply the retry policy. + if let Some(backoff) = retry_state.should_retry(&err, err.retries()) { + retry_state.sleep_backoff(backoff).await; + continue 'retry; + } + return if retry_state.total_retries() > 0 { + let retries = retry_state.total_retries(); + Err(err.into_retried(retries, retry_state.duration()?)) + } else { + Err(err) + }; + } + } + } + unreachable!("urls must not be empty"); + } +} + /// Whether the error is a status code error that is retryable. /// /// Port of `reqwest_retry::default_on_request_success`. diff --git a/crates/uv-client/src/lib.rs b/crates/uv-client/src/lib.rs index 594e935c9..5f7dea438 100644 --- a/crates/uv-client/src/lib.rs +++ b/crates/uv-client/src/lib.rs @@ -1,8 +1,9 @@ pub use base_client::{ AuthIntegration, BaseClient, BaseClientBuilder, DEFAULT_CONNECT_TIMEOUT, DEFAULT_MAX_REDIRECTS, DEFAULT_READ_TIMEOUT, DEFAULT_READ_TIMEOUT_UPLOAD, DEFAULT_RETRIES, ExtraMiddleware, - RedirectClientWithMiddleware, RedirectPolicy, RequestBuilder, RetryParsingError, RetryState, - UvRetryableStrategy, retryable_on_request_failure, + RedirectClientWithMiddleware, RedirectPolicy, RequestBuilder, RetriableError, + RetryParsingError, RetryState, UvRetryableStrategy, fetch_with_url_fallback, + retryable_on_request_failure, }; pub use cached_client::{CacheControl, CachedClient, CachedClientError, DataWithCachePolicy}; pub use error::{Error, ErrorKind, ProblemDetails, WrappedReqwestError}; diff --git a/crates/uv-python/src/downloads.rs b/crates/uv-python/src/downloads.rs index f20354e3d..6e6e32b81 100644 --- a/crates/uv-python/src/downloads.rs +++ b/crates/uv-python/src/downloads.rs @@ -18,10 +18,13 @@ use thiserror::Error; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, BufWriter, ReadBuf}; use tokio_util::compat::FuturesAsyncReadCompatExt; use tokio_util::either::Either; -use tracing::{debug, instrument, warn}; +use tracing::{debug, instrument}; use url::Url; -use uv_client::{BaseClient, RetryState, WrappedReqwestError, retryable_on_request_failure}; +use uv_client::{ + BaseClient, RetriableError, WrappedReqwestError, fetch_with_url_fallback, + retryable_on_request_failure, +}; use uv_distribution_filename::{ExtensionError, SourceDistExtension}; use uv_extract::hash::Hasher; use uv_fs::{Simplified, rename_with_retry}; @@ -126,7 +129,7 @@ pub enum Error { SystemTime(#[from] SystemTimeError), } -impl Error { +impl RetriableError for Error { // Return the number of retries that were made to complete this request before this error was // returned. // @@ -168,6 +171,14 @@ impl Error { _ => false, } } + + fn into_retried(self, retries: u32, duration: Duration) -> Self { + Self::NetworkErrorWithRetries { + err: Box::new(self), + retries, + duration, + } + } } /// The URL prefix used by `python-build-standalone` releases on GitHub. @@ -1152,55 +1163,20 @@ impl ManagedPythonDownload { reporter: Option<&dyn Reporter>, ) -> Result { let urls = self.download_urls(python_install_mirror, pypy_install_mirror)?; - let mut retry_state = RetryState::start( - *retry_policy, - // We pass in the last URL because that will trigger backoff if it fails. - urls.last().ok_or(Error::NoPythonDownloadUrlFound)?.clone(), - ); - - 'retry: loop { - for (i, url) in urls.iter().enumerate() { - let is_last = i == urls.len() - 1; - match self - .fetch_from_url( - url, - client, - installation_dir, - scratch_dir, - reinstall, - reporter, - ) - .await - { - Ok(download_result) => return Ok(download_result), - Err(err) => { - if !is_last && err.should_try_next_url() { - warn!( - "Failed to download `{}` from {url} ({err}); falling back to {}", - self.key(), - urls[i + 1] - ); - continue; - } - // All URLs exhausted; apply the retry policy. - if let Some(backoff) = retry_state.should_retry(&err, err.retries()) { - retry_state.sleep_backoff(backoff).await; - continue 'retry; - } - return if retry_state.total_retries() > 0 { - Err(Error::NetworkErrorWithRetries { - err: Box::new(err), - retries: retry_state.total_retries(), - duration: retry_state.duration()?, - }) - } else { - Err(err) - }; - } - } - } - unreachable!("download_urls() must return at least one URL"); + if urls.is_empty() { + return Err(Error::NoPythonDownloadUrlFound); } + fetch_with_url_fallback(&urls, *retry_policy, &format!("`{}`", self.key()), |url| { + self.fetch_from_url( + url, + client, + installation_dir, + scratch_dir, + reinstall, + reporter, + ) + }) + .await } /// Download and extract a Python distribution. @@ -1216,7 +1192,10 @@ impl ManagedPythonDownload { reporter: Option<&dyn Reporter>, ) -> Result { let urls = self.download_urls(python_install_mirror, pypy_install_mirror)?; - let url = urls.first().ok_or(Error::NoPythonDownloadUrlFound)?; + let url = urls + .into_iter() + .next() + .ok_or(Error::NoPythonDownloadUrlFound)?; self.fetch_from_url( url, client, @@ -1231,7 +1210,7 @@ impl ManagedPythonDownload { /// Download and extract a Python distribution from the given URL. async fn fetch_from_url( &self, - url: &DisplaySafeUrl, + url: DisplaySafeUrl, client: &BaseClient, installation_dir: &Path, scratch_dir: &Path, @@ -1303,7 +1282,7 @@ impl ManagedPythonDownload { } self.download_archive( - url, + &url, client, reporter, &python_builds_dir, @@ -1339,7 +1318,7 @@ impl ManagedPythonDownload { temp_dir.path().simplified_display() ); - let (reader, size) = read_url(url, client).await?; + let (reader, size) = read_url(&url, client).await?; self.extract_reader( reader, temp_dir.path(),