diff --git a/src-tauri/src/feat/profile.rs b/src-tauri/src/feat/profile.rs index bbaa5e8b7..677307990 100644 --- a/src-tauri/src/feat/profile.rs +++ b/src-tauri/src/feat/profile.rs @@ -101,7 +101,7 @@ async fn should_update_profile(uid: &String, ignore_auto_update: bool) -> Result } } -#[tracing::instrument(skip_all, level = "info", fields(uid = %uid, strategy = tracing::field::Empty))] +#[tracing::instrument(skip_all, level = "info", fields(uid = %uid, strategy = tracing::field::Empty, configured_elapsed = tracing::field::Empty, clash_elapsed = tracing::field::Empty, system_elapsed = tracing::field::Empty))] async fn perform_profile_update( uid: &String, url: &String, @@ -109,7 +109,7 @@ async fn perform_profile_update( option: Option<&PrfOption>, is_mannual_trigger: bool, ) -> Result<()> { - let mut merged_opt = PrfOption::merge(opt, option); + let merged_opt = PrfOption::merge(opt, option); let profiles = Config::profiles().await; let profiles_arc = profiles.latest_arc(); let profile_name = profiles_arc @@ -118,61 +118,74 @@ async fn perform_profile_update( .and_then(|item| item.name.clone()) .unwrap_or_else(|| String::from("UnKnown Profile")); - match PrfItem::from_url(url, None, None, merged_opt.as_ref()).await { - Ok(mut item) => { - logging!(info, Type::Config, "[订阅更新] 更新订阅配置成功"); - profiles_update_item_safe(uid, &mut item).await?; - return Ok(()); - } - Err(err) => { - logging!( - warn, - Type::Config, - "[订阅更新] 直接更新失败: {}，尝试使用Clash代理更新", - mask_err(&format!("{err:#}")) - ); - } + #[derive(Clone, Copy)] + enum Strategy { + Configured, + Clash, + System, } - merged_opt.get_or_insert_with(PrfOption::default).self_proxy = Some(true); - merged_opt.get_or_insert_with(PrfOption::default).with_proxy = Some(false); - - match PrfItem::from_url(url, None, None, merged_opt.as_ref()).await { - Ok(mut item) => { - logging!(info, Type::Config, "[订阅更新] 使用 Clash代理 更新订阅配置成功"); + let result = + crate::utils::retry::try_strategies(Strategy::Configured, [Strategy::Clash, Strategy::System], |strategy| { + let mut options = merged_opt.clone(); + async move { + let (name, success, failure) = match strategy { + Strategy::Configured => ( + "configured", + "[订阅更新] 更新订阅配置成功", + "[订阅更新] 直接更新失败: {}，尝试使用Clash代理更新", + ), + Strategy::Clash => ( + "clash", + "[订阅更新] 使用 Clash代理 更新订阅配置成功", + "[订阅更新] Clash代理更新失败: {}，尝试使用系统代理更新", + ), + Strategy::System => ( + "system", + "[订阅更新] 使用 系统代理 更新订阅配置成功", + "[订阅更新] 系统代理更新失败: {}，所有重试均已失败", + ), + }; + match strategy { + Strategy::Configured => {} + Strategy::Clash | Strategy::System => { + let option = options.get_or_insert_with(PrfOption::default); + option.self_proxy = Some(matches!(strategy, Strategy::Clash)); + option.with_proxy = Some(matches!(strategy, Strategy::System)); + } + } + tracing::Span::current().record("strategy", name); + let started = std::time::Instant::now(); + let result = PrfItem::from_url(url, None, None, options.as_ref()).await; + let elapsed_field = match strategy { + Strategy::Configured => "configured_elapsed", + Strategy::Clash => "clash_elapsed", + Strategy::System => "system_elapsed", + }; + tracing::Span::current().record(elapsed_field, tracing::field::debug(started.elapsed())); + match &result { + Ok(_) => logging!(info, Type::Config, "{success}"), + Err(error) => logging!( + warn, + Type::Config, + "{}", + failure.replace("{}", &mask_err(&format!("{error:#}"))) + ), + } + result + } + }) + .await; + let last_err = match result { + Ok((strategy, mut item)) => { + // A persistence error must not replay a successful download with another transport. profiles_update_item_safe(uid, &mut item).await?; - handle::Handle::notice(NoticeStatus::UpdateWithClashProxy, profile_name.as_str()); + if !matches!(strategy, Strategy::Configured) { + handle::Handle::notice(NoticeStatus::UpdateWithClashProxy, profile_name.as_str()); + } return Ok(()); } - Err(err) => { - logging!( - warn, - Type::Config, - "[订阅更新] Clash代理更新失败: {}，尝试使用系统代理更新", - mask_err(&format!("{err:#}")) - ); - } - } - - merged_opt.get_or_insert_with(PrfOption::default).self_proxy = Some(false); - merged_opt.get_or_insert_with(PrfOption::default).with_proxy = Some(true); - - let last_err = match PrfItem::from_url(url, None, None, merged_opt.as_ref()).await { - Ok(mut item) => { - logging!(info, Type::Config, "[订阅更新] 使用 系统代理 更新订阅配置成功"); - profiles_update_item_safe(uid, &mut item).await?; - handle::Handle::notice(NoticeStatus::UpdateWithClashProxy, profile_name.as_str()); - return Ok(()); - } - Err(err) => { - logging!( - warn, - Type::Config, - "[订阅更新] 系统代理更新失败: {}，所有重试均已失败", - mask_err(&format!("{err:#}")) - ); - err - } + Err(error) => error, }; let last_err = mask_err(&last_err.to_string()); diff --git a/src-tauri/src/utils/mod.rs b/src-tauri/src/utils/mod.rs index 106ec4669..e190227cb 100644 --- a/src-tauri/src/utils/mod.rs +++ b/src-tauri/src/utils/mod.rs @@ -11,6 +11,7 @@ pub(crate) mod network; pub use crate::core::notify::desktop as notification; pub mod port; pub mod resolve; +pub mod retry; #[cfg(target_os = "windows")] pub mod schtasks; pub mod server; diff --git a/src-tauri/src/utils/retry.rs b/src-tauri/src/utils/retry.rs new file mode 100644 index 000000000..280e2ab57 --- /dev/null +++ b/src-tauri/src/utils/retry.rs @@ -0,0 +1,153 @@ +use backon::{BackoffBuilder as _, ExponentialBackoff, ExponentialBuilder}; +use std::{future::Future, num::NonZeroUsize, time::Duration}; + +#[derive(Clone, Copy, Debug)] +pub enum Backoff { + Fixed(Duration), + #[allow(dead_code)] + Exponential { + base: Duration, + cap: Duration, + }, +} + +#[derive(Clone, Copy, Debug)] +pub struct RetryPolicy { + pub attempts: NonZeroUsize, + pub backoff: Backoff, + pub jitter: bool, +} + +impl RetryPolicy { + fn delays(self) -> ExponentialBackoff { + let (base, cap, factor) = match self.backoff { + Backoff::Fixed(delay) => (delay, delay, 1.0), + Backoff::Exponential { base, cap } => (base, cap, 2.0), + }; + let mut builder = ExponentialBuilder::default() + .with_min_delay(base) + .with_max_delay(cap) + .with_factor(factor) + .with_max_times(self.attempts.get() - 1); + if self.jitter { + builder = builder.with_jitter(); + } + builder.build() + } +} + +pub enum RetryError { + Retry(E), + #[allow(dead_code)] + Stop(E), +} + +pub async fn retry>>>( + policy: RetryPolicy, + mut operation: impl FnMut(usize) -> Fut, +) -> Result { + let mut result = operation(0).await; + for (index, delay) in policy.delays().enumerate() { + match result { + Ok(_) | Err(RetryError::Stop(_)) => break, + Err(RetryError::Retry(_)) => { + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + } + } + result = operation(index + 1).await; + } + result.map_err(|error| match error { + RetryError::Retry(error) | RetryError::Stop(error) => error, + }) +} + +pub async fn try_strategies>, const N: usize>( + first: S, + rest: [S; N], + mut operation: impl FnMut(S) -> Fut, +) -> Result<(S, T), E> { + retry( + RetryPolicy { + attempts: NonZeroUsize::MIN.saturating_add(N), + backoff: Backoff::Fixed(Duration::ZERO), + jitter: false, + }, + |attempt| { + let strategy = if attempt == 0 { first } else { rest[attempt - 1] }; + let future = operation(strategy); + async move { future.await.map(|value| (strategy, value)).map_err(RetryError::Retry) } + }, + ) + .await +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn retries_stop_at_success_terminal_error_or_attempt_limit() { + let policy = RetryPolicy { + attempts: NonZeroUsize::new(3).unwrap(), + backoff: Backoff::Fixed(Duration::ZERO), + jitter: false, + }; + let mut calls = 0; + let success = retry(policy, |attempt| { + calls += 1; + async move { + if attempt == 1 { + Ok(7) + } else { + Err(RetryError::Retry(attempt)) + } + } + }) + .await; + assert_eq!((success, calls), (Ok(7), 2)); + calls = 0; + let stopped = retry(policy, |_| { + calls += 1; + async move { Err::<(), _>(RetryError::Stop("terminal")) } + }) + .await; + assert_eq!((stopped, calls), (Err("terminal"), 1)); + calls = 0; + let exhausted = retry(policy, |attempt| { + calls += 1; + async move { Err::<(), _>(RetryError::Retry(attempt)) } + }) + .await; + assert_eq!((exhausted, calls), (Err(2), 3)); + } + + #[tokio::test] + async fn strategies_preserve_order_and_the_last_failure() { + let mut visited = Vec::new(); + let result = try_strategies("direct", ["clash", "system"], |strategy| { + visited.push(strategy); + async move { Err::<(), _>(strategy) } + }) + .await; + assert_eq!(result, Err("system")); + assert_eq!(visited, ["direct", "clash", "system"]); + } + + #[test] + fn backoff_keeps_the_attempt_budget_and_exponential_cap() { + let policy = RetryPolicy { + attempts: NonZeroUsize::new(5).unwrap(), + backoff: Backoff::Exponential { + base: Duration::from_millis(10), + cap: Duration::from_millis(25), + }, + jitter: false, + }; + assert_eq!( + policy.delays().collect::>(), + [10, 20, 25, 25].map(Duration::from_millis) + ); + } +}