diff --git a/crates/clickhousectl/src/error.rs b/crates/clickhousectl/src/error.rs index 8ab93232..893fd98d 100644 --- a/crates/clickhousectl/src/error.rs +++ b/crates/clickhousectl/src/error.rs @@ -54,6 +54,21 @@ pub enum Error { #[error("Download failed: {0}")] Download(String), + #[error("{0}")] + VersionNetwork(#[from] crate::version_manager::network::NetworkFailure), + + #[error("{fallback}; initial probe failure: {probe}")] + VersionFallback { + probe: crate::version_manager::network::NetworkFailure, + fallback: Box, + }, + + #[error("{failure} after {attempts} attempts")] + VersionNetworkRetryExhausted { + failure: crate::version_manager::network::NetworkFailure, + attempts: usize, + }, + #[error("No matching version found for: {0}")] NoMatchingVersion(String), diff --git a/crates/clickhousectl/src/version_manager/download.rs b/crates/clickhousectl/src/version_manager/download.rs index 1563a01f..307e2487 100644 --- a/crates/clickhousectl/src/version_manager/download.rs +++ b/crates/clickhousectl/src/version_manager/download.rs @@ -1,10 +1,38 @@ use crate::error::{Error, Result}; +use crate::version_manager::network::{NetworkFailure, NetworkStage, OperationClient}; use crate::version_manager::platform::{DownloadSource, Platform}; use futures_util::StreamExt; use indicatif::{ProgressBar, ProgressStyle}; use std::path::Path; +use std::time::Duration; use tokio::io::AsyncWriteExt; +#[derive(Debug, Clone, Copy)] +struct RetryPolicy { + max_attempts: usize, + initial_delay: Duration, + max_delay: Duration, +} + +impl RetryPolicy { + const INSTALLER: Self = Self { + max_attempts: 3, + initial_delay: Duration::from_millis(250), + max_delay: Duration::from_secs(5), + }; + + fn delay(self, failure: &NetworkFailure, attempt: usize) -> Duration { + failure + .retry_after + .unwrap_or_else(|| { + let exponent = u32::try_from(attempt.saturating_sub(1)).unwrap_or(u32::MAX); + self.initial_delay + .saturating_mul(2_u32.saturating_pow(exponent)) + }) + .min(self.max_delay) + } +} + /// Downloads from a DownloadSource to the specified path pub async fn download_from_source( source: &DownloadSource, @@ -17,14 +45,39 @@ pub async fn download_from_source( /// Downloads a file from a URL to the specified path, with progress bar pub async fn download_url(url: &str, dest_path: &Path) -> Result<()> { - let client = crate::http::client_builder().build()?; + let client = OperationClient::download(url)?; + download_url_with(&client, url, dest_path, RetryPolicy::INSTALLER).await +} + +async fn download_url_with( + client: &OperationClient, + url: &str, + dest_path: &Path, + policy: RetryPolicy, +) -> Result<()> { + for attempt in 1..=policy.max_attempts { + match download_once(client, url, dest_path).await { + Ok(()) => return Ok(()), + Err(Error::VersionNetwork(failure)) if failure.is_retryable() => { + if attempt == policy.max_attempts { + return Err(Error::VersionNetworkRetryExhausted { + failure, + attempts: attempt, + }); + } + client.sleep(policy.delay(&failure, attempt)).await; + } + Err(error) => return Err(error), + } + } + unreachable!("download retry policy always has at least one attempt") +} - let response = client - .get(url) - .send() - .await? - .error_for_status() - .map_err(|e| Error::Download(format!("Failed to download {}: {}", url, e)))?; +async fn download_once(client: &OperationClient, url: &str, dest_path: &Path) -> Result<()> { + let response = client.get(url, NetworkStage::Download).await?; + if !response.status().is_success() { + return Err(NetworkFailure::from_response(NetworkStage::Download, url, &response).into()); + } let total_size = response.content_length().unwrap_or(0); @@ -41,7 +94,15 @@ pub async fn download_url(url: &str, dest_path: &Path) -> Result<()> { let mut stream = response.bytes_stream(); while let Some(chunk) = stream.next().await { - let chunk = chunk?; + let chunk = match chunk { + Ok(chunk) => chunk, + Err(error) => { + pb.abandon(); + return Err( + NetworkFailure::from_request(NetworkStage::Download, url, &error).into(), + ); + } + }; file.write_all(&chunk).await?; downloaded += chunk.len() as u64; pb.set_position(downloaded); @@ -52,3 +113,184 @@ pub async fn download_url(url: &str, dest_path: &Path) -> Result<()> { pb.finish_with_message("Download complete"); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::version_manager::network::{NetworkCategory, Timeouts}; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + use wiremock::matchers::method; + use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; + + fn test_client(read: Duration, total: Duration) -> OperationClient { + OperationClient::with_timeouts(Timeouts::new(Duration::from_millis(100), read, total)) + .unwrap() + } + + fn test_policy(max_attempts: usize, max_delay: Duration) -> RetryPolicy { + RetryPolicy { + max_attempts, + initial_delay: Duration::ZERO, + max_delay, + } + } + + #[tokio::test] + async fn stalled_body_hits_read_deadline_and_exhausts_retries() { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + let address = listener.local_addr().unwrap(); + let requests = Arc::new(AtomicUsize::new(0)); + let server_requests = Arc::clone(&requests); + tokio::spawn(async move { + for _ in 0..2 { + let (mut socket, _) = listener.accept().await.unwrap(); + server_requests.fetch_add(1, Ordering::SeqCst); + tokio::spawn(async move { + let mut request = [0; 1024]; + let _ = socket.read(&mut request).await; + socket + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 10\r\nConnection: close\r\n\r\nabc", + ) + .await + .unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + }); + } + }); + let url = format!("http://{address}/binary"); + let temp = tempfile::tempdir().unwrap(); + let destination = temp.path().join("clickhouse"); + + let error = download_url_with( + &test_client(Duration::from_millis(40), Duration::from_millis(500)), + &url, + &destination, + test_policy(2, Duration::ZERO), + ) + .await + .unwrap_err(); + + assert!(matches!( + error, + Error::VersionNetworkRetryExhausted { + failure: NetworkFailure { + stage: NetworkStage::Download, + category: NetworkCategory::Timeout, + .. + }, + attempts: 2, + } + )); + assert_eq!(requests.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn retryable_server_errors_stop_after_three_attempts() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(503)) + .expect(3) + .mount(&server) + .await; + let temp = tempfile::tempdir().unwrap(); + + let error = download_url_with( + &test_client(Duration::from_millis(100), Duration::from_secs(2)), + &server.uri(), + &temp.path().join("clickhouse"), + test_policy(3, Duration::ZERO), + ) + .await + .unwrap_err(); + + assert!(matches!( + error, + Error::VersionNetworkRetryExhausted { + failure: NetworkFailure { + category: NetworkCategory::Server, + .. + }, + attempts: 3, + } + )); + } + + #[tokio::test] + async fn request_timeout_responses_are_classified_and_retried() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(408)) + .expect(3) + .mount(&server) + .await; + let temp = tempfile::tempdir().unwrap(); + + let error = download_url_with( + &test_client(Duration::from_millis(100), Duration::from_secs(2)), + &server.uri(), + &temp.path().join("clickhouse"), + test_policy(3, Duration::ZERO), + ) + .await + .unwrap_err(); + + assert!(matches!( + error, + Error::VersionNetworkRetryExhausted { + failure: NetworkFailure { + category: NetworkCategory::Timeout, + .. + }, + attempts: 3, + } + )); + } + + #[derive(Clone)] + struct RateLimitThenSuccess { + calls: Arc, + } + + impl Respond for RateLimitThenSuccess { + fn respond(&self, _request: &Request) -> ResponseTemplate { + if self.calls.fetch_add(1, Ordering::SeqCst) == 0 { + ResponseTemplate::new(429).insert_header("Retry-After", "60") + } else { + ResponseTemplate::new(200).set_body_bytes(b"complete".to_vec()) + } + } + } + + #[tokio::test] + async fn retry_after_is_honored_but_capped() { + let server = MockServer::start().await; + let calls = Arc::new(AtomicUsize::new(0)); + Mock::given(method("GET")) + .respond_with(RateLimitThenSuccess { + calls: Arc::clone(&calls), + }) + .mount(&server) + .await; + let temp = tempfile::tempdir().unwrap(); + let destination = temp.path().join("clickhouse"); + let max_delay = Duration::from_millis(30); + let started = tokio::time::Instant::now(); + + download_url_with( + &test_client(Duration::from_millis(100), Duration::from_secs(2)), + &server.uri(), + &destination, + test_policy(2, max_delay), + ) + .await + .unwrap(); + + assert!(started.elapsed() >= max_delay); + assert_eq!(calls.load(Ordering::SeqCst), 2); + assert_eq!(tokio::fs::read(destination).await.unwrap(), b"complete"); + } +} diff --git a/crates/clickhousectl/src/version_manager/list.rs b/crates/clickhousectl/src/version_manager/list.rs index 98f20d16..68e5b989 100644 --- a/crates/clickhousectl/src/version_manager/list.rs +++ b/crates/clickhousectl/src/version_manager/list.rs @@ -1,5 +1,6 @@ use crate::error::{Error, Result}; use crate::paths; +use crate::version_manager::network::{NetworkFailure, NetworkStage, OperationClient}; use chrono::Datelike; use serde::Deserialize; use std::fmt; @@ -69,18 +70,20 @@ pub struct VersionEntry { pub channel: Channel, } -/// Fetches available versions from GitHub releases -pub async fn list_available_versions() -> Result> { - let url = "https://api.github.com/repos/ClickHouse/ClickHouse/releases?per_page=100"; - let client = crate::http::client_builder().build()?; - - let response = client - .get(url) - .send() - .await? - .error_for_status() - .map_err(|e| Error::Download(format!("GitHub API request failed: {}", e)))?; - let releases: Vec = response.json().await?; +pub(crate) async fn list_available_versions_with( + client: &OperationClient, + url: &str, +) -> Result> { + let response = client.get(url, NetworkStage::VersionList).await?; + if !response.status().is_success() { + return Err( + NetworkFailure::from_response(NetworkStage::VersionList, url, &response).into(), + ); + } + let releases: Vec = response + .json() + .await + .map_err(|error| NetworkFailure::from_request(NetworkStage::VersionList, url, &error))?; let mut versions = Vec::new(); for release in releases { @@ -112,9 +115,8 @@ pub async fn list_available_versions_from_builds() -> Result> { use crate::version_manager::platform::{Platform, builds_probe_url}; let platform = Platform::detect()?; - let client = crate::http::client_builder() - .build() - .map_err(|e| Error::Download(e.to_string()))?; + let first_url = builds_probe_url("20.1", &platform); + let client = OperationClient::metadata(NetworkStage::BuildsList, &first_url)?; let current_year = chrono::Utc::now().year() as u32; // ClickHouse uses YY.MM versioning — scan from current year down to 20 (2020) @@ -127,11 +129,20 @@ pub async fn list_available_versions_from_builds() -> Result> { for mm in (1..=12).rev() { let version_path = format!("{}.{}", yy, mm); let url = builds_probe_url(&version_path, &platform); - match client.head(&url).send().await { - Ok(resp) if resp.status().is_success() => { + match client.head(&url, NetworkStage::BuildsList).await { + Ok(response) if response.status().is_success() => { available.push(version_path); } - _ => {} + Ok(response) if matches!(response.status().as_u16(), 403 | 404) => {} + Ok(response) => { + return Err(NetworkFailure::from_response( + NetworkStage::BuildsList, + &url, + &response, + ) + .into()); + } + Err(error) => return Err(error.into()), } } } diff --git a/crates/clickhousectl/src/version_manager/master.rs b/crates/clickhousectl/src/version_manager/master.rs index e84cd2f6..8ce7978f 100644 --- a/crates/clickhousectl/src/version_manager/master.rs +++ b/crates/clickhousectl/src/version_manager/master.rs @@ -14,6 +14,7 @@ use crate::error::Result; use crate::paths; +use crate::version_manager::network::{NetworkFailure, NetworkStage, OperationClient}; use crate::version_manager::platform::{DownloadSource, Platform}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; @@ -123,14 +124,21 @@ pub async fn head_info(platform: &Platform) -> Option { } .url(platform); - let client = reqwest::Client::builder() - .user_agent(crate::user_agent::user_agent()) - .build() - .ok()?; + let client = OperationClient::metadata(NetworkStage::MasterCheck, &url).ok()?; + head_info_from_url(&client, &url).await.ok().flatten() +} - let resp = client.head(&url).send().await.ok()?; +async fn head_info_from_url( + client: &OperationClient, + url: &str, +) -> std::result::Result, NetworkFailure> { + let resp = client.head(url, NetworkStage::MasterCheck).await?; if !resp.status().is_success() { - return None; + return Err(NetworkFailure::from_response( + NetworkStage::MasterCheck, + url, + &resp, + )); } let header = |name: reqwest::header::HeaderName| { resp.headers() @@ -138,12 +146,14 @@ pub async fn head_info(platform: &Platform) -> Option { .and_then(|v| v.to_str().ok()) .map(|s| s.to_string()) }; - let etag = header(reqwest::header::ETAG)?; + let Some(etag) = header(reqwest::header::ETAG) else { + return Ok(None); + }; let last_modified = header(reqwest::header::LAST_MODIFIED); - Some(HeadInfo { + Ok(Some(HeadInfo { etag, last_modified, - }) + })) } /// Pure reuse decision: reuse the recorded build only when we have a record, diff --git a/crates/clickhousectl/src/version_manager/mod.rs b/crates/clickhousectl/src/version_manager/mod.rs index 457083c0..b824f85f 100644 --- a/crates/clickhousectl/src/version_manager/mod.rs +++ b/crates/clickhousectl/src/version_manager/mod.rs @@ -2,6 +2,7 @@ pub mod download; pub mod install; pub mod list; pub mod master; +pub(crate) mod network; pub mod platform; pub mod resolve; pub mod spec; diff --git a/crates/clickhousectl/src/version_manager/network.rs b/crates/clickhousectl/src/version_manager/network.rs new file mode 100644 index 00000000..001fa5b4 --- /dev/null +++ b/crates/clickhousectl/src/version_manager/network.rs @@ -0,0 +1,345 @@ +use chrono::{DateTime, NaiveDateTime, Utc}; +use reqwest::header::RETRY_AFTER; +use std::fmt; +use std::time::Duration; +use thiserror::Error; +use tokio::time::Instant; + +const METADATA_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const METADATA_READ_TIMEOUT: Duration = Duration::from_secs(10); +const METADATA_TOTAL_TIMEOUT: Duration = Duration::from_secs(30); +const DOWNLOAD_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +const DOWNLOAD_READ_TIMEOUT: Duration = Duration::from_secs(30); +const DOWNLOAD_TOTAL_TIMEOUT: Duration = Duration::from_secs(30 * 60); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum NetworkStage { + BuildProbe, + BuildsList, + GithubLookup, + VersionList, + MasterCheck, + Download, +} + +impl fmt::Display for NetworkStage { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let name = match self { + Self::BuildProbe => "build probe", + Self::BuildsList => "builds list", + Self::GithubLookup => "GitHub version lookup", + Self::VersionList => "GitHub version list", + Self::MasterCheck => "master check", + Self::Download => "download", + }; + f.write_str(name) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum NetworkCategory { + Timeout, + Connect, + Transport, + InvalidResponse, + Forbidden, + NotFound, + RateLimited, + Server, + Client, + UnexpectedStatus, +} + +impl fmt::Display for NetworkCategory { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let category = match self { + Self::Timeout => "timeout", + Self::Connect => "connection error", + Self::Transport => "transport error", + Self::InvalidResponse => "invalid response", + Self::Forbidden => "forbidden (HTTP 403)", + Self::NotFound => "not found (HTTP 404)", + Self::RateLimited => "rate limited (HTTP 429)", + Self::Server => "server error (HTTP 5xx)", + Self::Client => "client error (HTTP 4xx)", + Self::UnexpectedStatus => "unexpected HTTP status", + }; + f.write_str(category) + } +} + +#[derive(Debug, Clone, Error)] +#[error("{stage} request to {host} failed: {category}")] +pub(crate) struct NetworkFailure { + pub(crate) stage: NetworkStage, + pub(crate) host: String, + pub(crate) category: NetworkCategory, + pub(crate) retry_after: Option, +} + +impl NetworkFailure { + pub(crate) fn from_request(stage: NetworkStage, url: &str, error: &reqwest::Error) -> Self { + let category = if error.is_timeout() { + NetworkCategory::Timeout + } else if error.is_connect() { + NetworkCategory::Connect + } else if error.is_decode() { + NetworkCategory::InvalidResponse + } else { + NetworkCategory::Transport + }; + Self::new(stage, url, category, None) + } + + pub(crate) fn from_response( + stage: NetworkStage, + url: &str, + response: &reqwest::Response, + ) -> Self { + let status = response.status(); + let category = match status.as_u16() { + 403 => NetworkCategory::Forbidden, + 404 => NetworkCategory::NotFound, + 408 => NetworkCategory::Timeout, + 429 => NetworkCategory::RateLimited, + _ if status.is_server_error() => NetworkCategory::Server, + _ if status.is_client_error() => NetworkCategory::Client, + _ => NetworkCategory::UnexpectedStatus, + }; + let retry_after = response + .headers() + .get(RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(parse_retry_after); + Self::new(stage, url, category, retry_after) + } + + pub(crate) fn timeout(stage: NetworkStage, url: &str) -> Self { + Self::new(stage, url, NetworkCategory::Timeout, None) + } + + pub(crate) fn is_retryable(&self) -> bool { + matches!( + self.category, + NetworkCategory::Timeout + | NetworkCategory::Connect + | NetworkCategory::Transport + | NetworkCategory::RateLimited + | NetworkCategory::Server + ) + } + + fn new( + stage: NetworkStage, + url: &str, + category: NetworkCategory, + retry_after: Option, + ) -> Self { + Self { + stage, + host: reqwest::Url::parse(url) + .ok() + .and_then(|url| url.host_str().map(str::to_owned)) + .unwrap_or_else(|| "unknown host".to_string()), + category, + retry_after, + } + } +} + +fn parse_retry_after(value: &str) -> Option { + if let Ok(seconds) = value.parse::() { + return Some(Duration::from_secs(seconds)); + } + + let retry_at = DateTime::parse_from_rfc2822(value) + .map(|date| date.with_timezone(&Utc)) + .or_else(|_| { + NaiveDateTime::parse_from_str(value, "%A, %d-%b-%y %H:%M:%S GMT") + .map(|date| date.and_utc()) + }) + .or_else(|_| { + NaiveDateTime::parse_from_str(value, "%a %b %e %H:%M:%S %Y").map(|date| date.and_utc()) + }) + .ok()?; + retry_at.signed_duration_since(Utc::now()).to_std().ok() +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct Timeouts { + connect: Duration, + read: Duration, + total: Duration, +} + +impl Timeouts { + const METADATA: Self = Self { + connect: METADATA_CONNECT_TIMEOUT, + read: METADATA_READ_TIMEOUT, + total: METADATA_TOTAL_TIMEOUT, + }; + const DOWNLOAD: Self = Self { + connect: DOWNLOAD_CONNECT_TIMEOUT, + read: DOWNLOAD_READ_TIMEOUT, + total: DOWNLOAD_TOTAL_TIMEOUT, + }; + + #[cfg(test)] + pub(crate) const fn new(connect: Duration, read: Duration, total: Duration) -> Self { + Self { + connect, + read, + total, + } + } +} + +#[derive(Clone)] +pub(crate) struct OperationClient { + inner: reqwest::Client, + deadline: Instant, +} + +impl OperationClient { + pub(crate) fn metadata(stage: NetworkStage, url: &str) -> Result { + Self::build(Timeouts::METADATA) + .map_err(|error| NetworkFailure::from_request(stage, url, &error)) + } + + pub(crate) fn download(url: &str) -> Result { + Self::build(Timeouts::DOWNLOAD) + .map_err(|error| NetworkFailure::from_request(NetworkStage::Download, url, &error)) + } + + fn build(timeouts: Timeouts) -> reqwest::Result { + let inner = crate::http::client_builder() + .connect_timeout(timeouts.connect) + .read_timeout(timeouts.read) + .timeout(timeouts.total) + .build()?; + Ok(Self { + inner, + deadline: Instant::now() + timeouts.total, + }) + } + + #[cfg(test)] + pub(crate) fn with_timeouts(timeouts: Timeouts) -> reqwest::Result { + Self::build(timeouts) + } + + pub(crate) async fn get( + &self, + url: &str, + stage: NetworkStage, + ) -> Result { + self.send(self.inner.get(url), url, stage).await + } + + pub(crate) async fn head( + &self, + url: &str, + stage: NetworkStage, + ) -> Result { + self.send(self.inner.head(url), url, stage).await + } + + async fn send( + &self, + request: reqwest::RequestBuilder, + url: &str, + stage: NetworkStage, + ) -> Result { + let remaining = self + .deadline + .checked_duration_since(Instant::now()) + .ok_or_else(|| NetworkFailure::timeout(stage, url))?; + request + .timeout(remaining) + .send() + .await + .map_err(|error| NetworkFailure::from_request(stage, url, &error)) + } + + pub(crate) async fn sleep(&self, duration: Duration) { + let wake_at = (Instant::now() + duration).min(self.deadline); + tokio::time::sleep_until(wake_at).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::net::TcpListener; + + #[test] + fn retry_after_seconds_are_parsed() { + assert_eq!(parse_retry_after("7"), Some(Duration::from_secs(7))); + } + + #[test] + fn retry_after_http_date_forms_are_parsed() { + for value in [ + "Sat, 06 Nov 2060 08:49:37 GMT", + "Saturday, 06-Nov-60 08:49:37 GMT", + "Sat Nov 6 08:49:37 2060", + ] { + assert!(parse_retry_after(value).is_some(), "{value}"); + } + } + + #[test] + fn expired_retry_after_dates_are_ignored() { + let past = Utc::now() - chrono::Duration::seconds(30); + for value in [ + past.format("%a, %d %b %Y %H:%M:%S GMT").to_string(), + past.format("%A, %d-%b-%y %H:%M:%S GMT").to_string(), + past.format("%a %b %e %H:%M:%S %Y").to_string(), + ] { + assert_eq!(parse_retry_after(&value), None, "{value}"); + } + } + + #[test] + fn errors_expose_only_stage_host_and_category() { + let failure = NetworkFailure::new( + NetworkStage::Download, + "https://user:secret@example.com/private?token=secret", + NetworkCategory::RateLimited, + None, + ); + + assert_eq!(failure.host, "example.com"); + assert_eq!( + failure.to_string(), + "download request to example.com failed: rate limited (HTTP 429)" + ); + assert!(!failure.to_string().contains("secret")); + } + + #[tokio::test] + async fn total_deadline_bounds_stalled_headers() { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (_socket, _) = listener.accept().await.unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + }); + let url = format!("http://{address}/stalled"); + let client = OperationClient::with_timeouts(Timeouts::new( + Duration::from_millis(100), + Duration::from_secs(1), + Duration::from_millis(50), + )) + .unwrap(); + + let error = client + .get(&url, NetworkStage::GithubLookup) + .await + .unwrap_err(); + + assert_eq!(error.stage, NetworkStage::GithubLookup); + assert_eq!(error.host, "127.0.0.1"); + assert_eq!(error.category, NetworkCategory::Timeout); + } +} diff --git a/crates/clickhousectl/src/version_manager/resolve.rs b/crates/clickhousectl/src/version_manager/resolve.rs index bef8d5ea..cd24a9a9 100644 --- a/crates/clickhousectl/src/version_manager/resolve.rs +++ b/crates/clickhousectl/src/version_manager/resolve.rs @@ -1,11 +1,15 @@ use crate::error::{Error, Result}; use crate::version_manager::list::{ - Channel, VersionEntry, list_available_versions, list_installed_versions, + Channel, VersionEntry, list_available_versions_with, list_installed_versions, }; +use crate::version_manager::network::{NetworkFailure, NetworkStage, OperationClient}; use crate::version_manager::platform::{DownloadSource, Platform, builds_probe_url}; use crate::version_manager::spec::VersionSpec; use serde::Deserialize; +const GITHUB_RELEASES_URL: &str = + "https://api.github.com/repos/ClickHouse/ClickHouse/releases?per_page=100"; + /// Result of resolving a version spec — contains everything needed to download #[derive(Debug, Clone)] pub struct ResolvedVersion { @@ -58,12 +62,32 @@ fn find_local_match(spec: &VersionSpec, installed: &[String]) -> Option /// Resolve a VersionSpec into a concrete download source pub async fn resolve(spec: &VersionSpec, platform: &Platform) -> Result { + if matches!(spec, VersionSpec::Latest) { + return resolve_latest(platform).await; + } + + let (stage, initial_url) = match spec { + VersionSpec::Channel(_) | VersionSpec::Exact(_) => { + (NetworkStage::GithubLookup, GITHUB_RELEASES_URL.to_string()) + } + VersionSpec::Major(major) => ( + NetworkStage::BuildProbe, + builds_probe_url(&format!("{}.1", major), platform), + ), + VersionSpec::Minor(major, minor) => ( + NetworkStage::BuildProbe, + builds_probe_url(&format!("{}.{}", major, minor), platform), + ), + VersionSpec::Latest => unreachable!(), + }; + let client = OperationClient::metadata(stage, &initial_url)?; + match spec { - VersionSpec::Latest => resolve_latest(platform).await, - VersionSpec::Channel(channel) => resolve_channel(*channel, platform).await, - VersionSpec::Major(major) => resolve_major(*major, platform).await, - VersionSpec::Minor(major, minor) => resolve_minor(*major, *minor, platform).await, - VersionSpec::Exact(version) => resolve_exact(version, platform).await, + VersionSpec::Latest => unreachable!(), + VersionSpec::Channel(channel) => resolve_channel(*channel, platform, &client).await, + VersionSpec::Major(major) => resolve_major(*major, platform, &client).await, + VersionSpec::Minor(major, minor) => resolve_minor(*major, *minor, platform, &client).await, + VersionSpec::Exact(version) => resolve_exact(version, platform, &client).await, } } @@ -81,8 +105,12 @@ async fn resolve_latest(_platform: &Platform) -> Result { } /// `install stable` / `install lts` — GH API to find minor, then builds -async fn resolve_channel(channel: Channel, platform: &Platform) -> Result { - let available = list_available_versions().await?; +async fn resolve_channel( + channel: Channel, + platform: &Platform, + client: &OperationClient, +) -> Result { + let available = list_available_versions_with(client, GITHUB_RELEASES_URL).await?; let entry = available .iter() .find(|e| e.channel == channel) @@ -92,7 +120,10 @@ async fn resolve_channel(channel: Channel, platform: &Platform) -> Result Result Result { +async fn resolve_major( + major: u32, + platform: &Platform, + client: &OperationClient, +) -> Result { // Probe builds.clickhouse.com for all possible minors in this major (1..12) let mut highest_available: Option = None; - let client = crate::http::client_builder() - .build() - .map_err(|e| Error::Download(e.to_string()))?; + let mut first_probe_failure = None; for minor in 1..=12 { let url = builds_probe_url(&format!("{}.{}", major, minor), platform); - match client.head(&url).send().await { - Ok(resp) if resp.status().is_success() => { - highest_available = Some(minor); + match probe_builds(client, &url).await { + Ok(ProbeOutcome::Available) => highest_available = Some(minor), + Ok(ProbeOutcome::Unavailable(_)) => {} + Err(error) => { + first_probe_failure = Some(error); + break; } - _ => {} } } - if let Some(minor) = highest_available { + if first_probe_failure.is_none() + && let Some(minor) = highest_available + { let version_path = format!("{}.{}", major, minor); return Ok(ResolvedVersion { source: DownloadSource::Builds { @@ -142,49 +179,82 @@ async fn resolve_major(major: u32, platform: &Platform) -> Result return Ok(fallback_source(&entry.version, entry.channel, platform)), + Err(Error::NoMatchingVersion(_)) => {} + Err(error) => return Err(preserve_probe_failure(first_probe_failure, error)), } } - Err(Error::NoMatchingVersion(major.to_string())) + Err(preserve_probe_failure( + first_probe_failure, + Error::NoMatchingVersion(major.to_string()), + )) } /// `install 25.12` — try builds, fallback to packages/GH -async fn resolve_minor(major: u32, minor: u32, platform: &Platform) -> Result { +async fn resolve_minor( + major: u32, + minor: u32, + platform: &Platform, + client: &OperationClient, +) -> Result { let version_path = format!("{}.{}", major, minor); + let build_url = builds_probe_url(&version_path, platform); + let refs_url = version_refs_url(&version_path); + resolve_minor_from_urls(client, &version_path, platform, &build_url, &refs_url).await +} +async fn resolve_minor_from_urls( + client: &OperationClient, + version_path: &str, + platform: &Platform, + build_url: &str, + refs_url: &str, +) -> Result { // Try builds first - if probe_builds(&version_path, platform).await { - return Ok(ResolvedVersion { - source: DownloadSource::Builds { - version_path: version_path.clone(), - }, - display_version: version_path, - exact_version_known: false, - exact_version: None, - channel: None, - }); - } + let probe_failure = match probe_builds(client, build_url).await { + Ok(ProbeOutcome::Available) => { + return Ok(ResolvedVersion { + source: DownloadSource::Builds { + version_path: version_path.to_string(), + }, + display_version: version_path.to_string(), + exact_version_known: false, + exact_version: None, + channel: None, + }); + } + Ok(ProbeOutcome::Unavailable(_)) => None, + Err(error) => Some(error), + }; // Fallback: targeted GH API call to find exact version for this minor - let entry = find_version_by_refs(&version_path).await?; + let entry = find_version_by_refs(client, version_path, refs_url) + .await + .map_err(|error| preserve_probe_failure(probe_failure, error))?; Ok(fallback_source(&entry.version, entry.channel, platform)) } /// `install 25.12.9.61` — exact version, needs channel from GH API -async fn resolve_exact(version: &str, platform: &Platform) -> Result { +async fn resolve_exact( + version: &str, + platform: &Platform, + client: &OperationClient, +) -> Result { // Use matching-refs to find the exact tag and its channel. // For "25.12.9.61", search refs matching "v25.12.9.61" — should return the exact tag. // Fail fast if the lookup fails: a wrong channel produces a broken download URL, // and silently guessing Stable could fetch the wrong artifact. - match find_exact_channel(version).await { + match find_exact_channel(client, version).await { Ok(channel) => Ok(fallback_source(version, channel, platform)), Err(Error::NoMatchingVersion(_)) => { let series = extract_minor(version)?; // The exact miss is definitive; fetching a retry hint is best-effort. - let available = find_version_by_refs(&series).await.ok(); + let url = version_refs_url(&series); + let available = find_version_by_refs(client, &series, &url).await.ok(); Err(exact_version_no_match(version, &series, available.as_ref())) } @@ -204,21 +274,22 @@ fn exact_version_no_match(version: &str, series: &str, available: Option<&Versio } /// Look up the channel for an exact version via GitHub's matching-refs API -async fn find_exact_channel(version: &str) -> Result { +async fn find_exact_channel(client: &OperationClient, version: &str) -> Result { let url = format!( "https://api.github.com/repos/ClickHouse/ClickHouse/git/matching-refs/tags/v{}-", version ); - let client = crate::http::client_builder().build()?; - - let response = client - .get(&url) - .send() - .await? - .error_for_status() - .map_err(|e| Error::Download(format!("GitHub API request failed: {}", e)))?; + let response = client.get(&url, NetworkStage::GithubLookup).await?; + if !response.status().is_success() { + return Err( + NetworkFailure::from_response(NetworkStage::GithubLookup, &url, &response).into(), + ); + } - let refs: Vec = response.json().await?; + let refs: Vec = response + .json() + .await + .map_err(|error| NetworkFailure::from_request(NetworkStage::GithubLookup, &url, &error))?; parse_exact_channel(&refs, version) } @@ -277,17 +348,34 @@ fn fallback_source(version: &str, channel: Channel, platform: &Platform) -> Reso } } -/// Probe builds.clickhouse.com with a HEAD request to check if a version exists -async fn probe_builds(version_path: &str, platform: &Platform) -> bool { - let url = builds_probe_url(version_path, platform); - let client = match crate::http::client_builder().build() { - Ok(c) => c, - Err(_) => return false, - }; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ProbeMiss { + Forbidden, + NotFound, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ProbeOutcome { + Available, + Unavailable(ProbeMiss), +} - match client.head(&url).send().await { - Ok(resp) => resp.status().is_success(), - Err(_) => false, +/// Probe builds.clickhouse.com with a HEAD request to check if a version exists. +/// Only 403/404 mean that fallback is expected; outages remain errors. +async fn probe_builds( + client: &OperationClient, + url: &str, +) -> std::result::Result { + let response = client.head(url, NetworkStage::BuildProbe).await?; + match response.status().as_u16() { + _ if response.status().is_success() => Ok(ProbeOutcome::Available), + 403 => Ok(ProbeOutcome::Unavailable(ProbeMiss::Forbidden)), + 404 => Ok(ProbeOutcome::Unavailable(ProbeMiss::NotFound)), + _ => Err(NetworkFailure::from_response( + NetworkStage::BuildProbe, + url, + &response, + )), } } @@ -300,24 +388,42 @@ struct GitRef { /// Find the latest release version matching a prefix using GitHub's matching-refs API. /// This is a single targeted API call that works regardless of how old the version is. /// prefix should be like "25.2" or "24.8" — we search for tags matching `v{prefix}.` -async fn find_version_by_refs(prefix: &str) -> Result { - let url = format!( +fn version_refs_url(prefix: &str) -> String { + format!( "https://api.github.com/repos/ClickHouse/ClickHouse/git/matching-refs/tags/v{}.", prefix - ); - let client = crate::http::client_builder().build()?; + ) +} - let response = client - .get(&url) - .send() - .await? - .error_for_status() - .map_err(|e| Error::Download(format!("GitHub API request failed: {}", e)))?; +async fn find_version_by_refs( + client: &OperationClient, + prefix: &str, + url: &str, +) -> Result { + let response = client.get(url, NetworkStage::GithubLookup).await?; + if !response.status().is_success() { + return Err( + NetworkFailure::from_response(NetworkStage::GithubLookup, url, &response).into(), + ); + } - let refs: Vec = response.json().await?; + let refs: Vec = response + .json() + .await + .map_err(|error| NetworkFailure::from_request(NetworkStage::GithubLookup, url, &error))?; parse_version_refs(&refs, prefix) } +fn preserve_probe_failure(probe: Option, fallback: Error) -> Error { + match probe { + Some(probe) => Error::VersionFallback { + probe, + fallback: Box::new(fallback), + }, + None => fallback, + } +} + /// Parse a list of git refs into the best matching VersionEntry. /// Prefers stable/lts tags, but falls back to any tagged version (e.g. "-new") /// so that pre-release or newly-tagged versions can still be resolved. @@ -374,7 +480,11 @@ fn extract_minor(version: &str) -> Result { #[cfg(test)] mod tests { use super::*; + use crate::version_manager::network::{NetworkCategory, Timeouts}; use crate::version_manager::platform::Os; + use std::time::Duration; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; fn make_ref(name: &str) -> GitRef { GitRef { @@ -382,6 +492,132 @@ mod tests { } } + fn test_client() -> OperationClient { + OperationClient::with_timeouts(Timeouts::new( + Duration::from_millis(100), + Duration::from_millis(100), + Duration::from_secs(2), + )) + .unwrap() + } + + fn macos_platform() -> Platform { + Platform { + os: Os::MacOS, + arch: crate::version_manager::platform::Arch::Aarch64, + } + } + + #[tokio::test] + async fn build_probe_distinguishes_403_and_404_as_expected_misses() { + for (status, expected) in [(403, ProbeMiss::Forbidden), (404, ProbeMiss::NotFound)] { + let server = MockServer::start().await; + Mock::given(method("HEAD")) + .respond_with(ResponseTemplate::new(status)) + .mount(&server) + .await; + + let outcome = probe_builds(&test_client(), &server.uri()).await.unwrap(); + assert_eq!(outcome, ProbeOutcome::Unavailable(expected)); + } + } + + #[tokio::test] + async fn build_probe_classifies_429_and_5xx_as_failures() { + for (status, expected) in [ + (429, NetworkCategory::RateLimited), + (503, NetworkCategory::Server), + ] { + let server = MockServer::start().await; + Mock::given(method("HEAD")) + .respond_with(ResponseTemplate::new(status)) + .mount(&server) + .await; + + let error = probe_builds(&test_client(), &server.uri()) + .await + .unwrap_err(); + assert_eq!(error.stage, NetworkStage::BuildProbe); + assert_eq!(error.category, expected); + } + } + + #[tokio::test] + async fn expected_probe_miss_falls_back_to_github() { + let server = MockServer::start().await; + Mock::given(method("HEAD")) + .and(path("/build")) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/refs")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([ + {"ref": "refs/tags/v25.12.9.61-stable"} + ]))) + .mount(&server) + .await; + + let resolved = resolve_minor_from_urls( + &test_client(), + "25.12", + &macos_platform(), + &format!("{}/build", server.uri()), + &format!("{}/refs", server.uri()), + ) + .await + .unwrap(); + + assert!(matches!( + resolved.source, + DownloadSource::GitHub { + ref version, + channel: Channel::Stable, + } if version == "25.12.9.61" + )); + } + + #[tokio::test] + async fn fallback_failure_preserves_original_probe_failure() { + let server = MockServer::start().await; + Mock::given(method("HEAD")) + .and(path("/build")) + .respond_with(ResponseTemplate::new(503)) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/refs")) + .respond_with(ResponseTemplate::new(429)) + .mount(&server) + .await; + + let error = resolve_minor_from_urls( + &test_client(), + "25.12", + &macos_platform(), + &format!("{}/build", server.uri()), + &format!("{}/refs", server.uri()), + ) + .await + .unwrap_err(); + + match error { + Error::VersionFallback { probe, fallback } => { + assert_eq!(probe.stage, NetworkStage::BuildProbe); + assert_eq!(probe.category, NetworkCategory::Server); + assert!(matches!( + *fallback, + Error::VersionNetwork(NetworkFailure { + stage: NetworkStage::GithubLookup, + category: NetworkCategory::RateLimited, + .. + }) + )); + } + other => panic!("expected preserved fallback error, got {other:?}"), + } + } + #[test] fn test_extract_minor() { assert_eq!(extract_minor("25.12.9.61").unwrap(), "25.12");