refactor(utils): unified retry and strategy-list helpers

This commit is contained in:
Tunglies
2026-10-01 14:34:16 +08:00
committed by GitHub
parent 7eb24d8c69
commit edcd76f27d
3 changed files with 219 additions and 52 deletions
+65 -52
View File
@@ -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());
+1
View File
@@ -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;
+153
View File
@@ -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<E> {
Retry(E),
#[allow(dead_code)]
Stop(E),
}
pub async fn retry<T, E, Fut: Future<Output = Result<T, RetryError<E>>>>(
policy: RetryPolicy,
mut operation: impl FnMut(usize) -> Fut,
) -> Result<T, E> {
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<S: Copy, T, E, Fut: Future<Output = Result<T, E>>, 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::<Vec<_>>(),
[10, 20, 25, 25].map(Duration::from_millis)
);
}
}