From dfc729a1705194916c11bfff61017bcb3b8c10c6 Mon Sep 17 00:00:00 2001 From: Tunglies <77394545+Tunglies@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:32:59 +0800 Subject: [PATCH] refactor(backend): flatten refresh publication and retry policies --- src-tauri/src/core/notify/announce.rs | 53 ++------------------------- src-tauri/src/core/notify/mod.rs | 2 +- src-tauri/src/feat/executor.rs | 18 +++------ src-tauri/src/utils/retry.rs | 48 +++--------------------- 4 files changed, 15 insertions(+), 106 deletions(-) diff --git a/src-tauri/src/core/notify/announce.rs b/src-tauri/src/core/notify/announce.rs index b6fb11234..9b727e681 100644 --- a/src-tauri/src/core/notify/announce.rs +++ b/src-tauri/src/core/notify/announce.rs @@ -1,7 +1,6 @@ use super::{handle::Handle, notification::FrontendEvent}; -use anyhow::Result; use parking_lot::Mutex; -use std::{cell::RefCell, future::Future, time::Duration}; +use std::time::Duration; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Refresh { @@ -11,10 +10,6 @@ pub enum Refresh { Proxies, } -tokio::task_local! { - static PENDING: RefCell>; -} - const REFRESH_WINDOW: Duration = Duration::from_millis(20); static BATCH: RefreshBatch = RefreshBatch(Mutex::new(Vec::new())); @@ -42,7 +37,7 @@ impl RefreshBatch { } pub fn announce(refresh: Refresh) { - if PENDING.try_with(|pending| pending.borrow_mut().push(refresh)).is_err() && BATCH.enqueue(refresh) { + if BATCH.enqueue(refresh) { crate::AsyncHandler::spawn(|| BATCH.flush(emit)); } } @@ -56,54 +51,12 @@ fn emit(refresh: Refresh) { }); } -async fn collect_committed(operation: impl Future> + Send) -> Result<(T, Vec)> { - PENDING - .scope(RefCell::new(Vec::new()), async { - let value = operation.await?; - let refreshes = PENDING.with(|pending| pending.take()); - Ok((value, refreshes)) - }) - .await -} - -pub async fn after_commit(operation: impl Future> + Send) -> Result { - let (value, refreshes) = collect_committed(operation).await?; - // Nested transactions hand their notifications to the outer transaction. - for refresh in refreshes { - announce(refresh); - } - Ok(value) -} - #[cfg(test)] mod tests { use super::*; #[tokio::test] - async fn only_successful_transactions_publish_refreshes() -> Result<()> { - let (_, refreshes) = collect_committed(async { - announce(Refresh::Clash); - after_commit(async { - announce(Refresh::Verge); - Ok(()) - }) - .await?; - let failed = after_commit(async { - announce(Refresh::Profiles); - Err::<(), _>(anyhow::anyhow!("persistence failed")) - }) - .await; - assert!(failed.is_err()); - announce(Refresh::Clash); - Ok(()) - }) - .await?; - assert_eq!(refreshes, [Refresh::Clash, Refresh::Verge, Refresh::Clash]); - Ok(()) - } - - #[tokio::test] - async fn bursts_keep_each_refresh_kind_and_rearm_during_delivery() -> Result<()> { + async fn bursts_keep_each_refresh_kind_and_rearm_during_delivery() -> anyhow::Result<()> { let batch = RefreshBatch::default(); let delivered = Mutex::new(Vec::new()); assert!(batch.enqueue(Refresh::Clash)); diff --git a/src-tauri/src/core/notify/mod.rs b/src-tauri/src/core/notify/mod.rs index cc59f0b36..55d3a43c3 100644 --- a/src-tauri/src/core/notify/mod.rs +++ b/src-tauri/src/core/notify/mod.rs @@ -1,5 +1,5 @@ mod announce; -pub use announce::{Refresh, after_commit, announce}; +pub use announce::{Refresh, announce}; pub mod desktop; pub mod handle; pub mod notification; diff --git a/src-tauri/src/feat/executor.rs b/src-tauri/src/feat/executor.rs index 471f265b7..c301a6975 100644 --- a/src-tauri/src/feat/executor.rs +++ b/src-tauri/src/feat/executor.rs @@ -63,8 +63,6 @@ async fn ensure_effect( } Effect::ClashConfig => { manager.update_config_in_patch(update).await?; - Config::runtime().await.apply(); - announce(Refresh::Clash); } Effect::Autostart => autostart::update_launch().await?, Effect::Language => { @@ -123,7 +121,7 @@ async fn ensure_effect( Ok(()) } -async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> { +pub(super) async fn apply(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> { let manager = CoreManager::global(); let update = manager.claim_config_update(config_write)?; let clash = Config::clash().await; @@ -167,6 +165,7 @@ async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effect if let Err(error) = result { transaction.rollback(); if !Arc::ptr_eq(&original_runtime, &runtime.data_arc()) { + announce(Refresh::Clash); // The Core already loaded this runtime; restoring its file would only hide that state. let runtime_path = crate::utils::dirs::app_home_dir()?.join(crate::constants::files::RUNTIME_CONFIG); snapshots.retain(|snapshot| snapshot.path != runtime_path); @@ -179,6 +178,9 @@ async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effect transaction.commit(); match patch { Patch::Verge { .. } => { + if effects.contains(&Effect::ClashConfig) { + announce(Refresh::Clash); + } logging_error!(Type::Backup, AutoBackupManager::global().refresh_settings().await); announce(Refresh::Verge); } @@ -187,16 +189,6 @@ async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effect Ok(()) } -pub(super) async fn apply(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> { - let runtime = Config::runtime().await; - let original = runtime.data_arc(); - let result = crate::core::notify::after_commit(apply_inner(config_write, patch, effects)).await; - if result.is_err() && !Arc::ptr_eq(&original, &runtime.data_arc()) { - announce(Refresh::Clash); - } - result -} - #[cfg(test)] mod tests { use super::*; diff --git a/src-tauri/src/utils/retry.rs b/src-tauri/src/utils/retry.rs index 5becccac1..72c79017b 100644 --- a/src-tauri/src/utils/retry.rs +++ b/src-tauri/src/utils/retry.rs @@ -1,46 +1,18 @@ -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, + delay: Duration, } impl RetryPolicy { pub const fn fixed(attempts: NonZeroUsize, delay: Duration) -> Self { - Self { - attempts, - backoff: Backoff::Fixed(delay), - jitter: false, - } + Self { attempts, delay } } - 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() + fn delays(self) -> impl Iterator { + std::iter::repeat_n(self.delay, self.attempts.get() - 1) } } @@ -82,11 +54,7 @@ pub async fn try_strategies< mut operation: impl FnMut(S) -> Fut + Send, ) -> Result<(S, T), E> { retry( - RetryPolicy { - attempts: NonZeroUsize::MIN.saturating_add(N), - backoff: Backoff::Fixed(Duration::ZERO), - jitter: false, - }, + RetryPolicy::fixed(NonZeroUsize::MIN.saturating_add(N), Duration::ZERO), move |attempt| { let strategy = if attempt == 0 { first } else { rest[attempt - 1] }; let future = operation(strategy); @@ -145,11 +113,7 @@ mod tests { #[tokio::test] async fn retries_stop_at_success_terminal_error_or_attempt_limit() { - let policy = RetryPolicy { - attempts: NonZeroUsize::MIN.saturating_add(2), - backoff: Backoff::Fixed(Duration::ZERO), - jitter: false, - }; + let policy = RetryPolicy::fixed(NonZeroUsize::MIN.saturating_add(2), Duration::ZERO); let mut calls = 0; let success = retry(policy, |attempt| { calls += 1;