refactor(backend): flatten refresh publication and retry policies

This commit is contained in:
Tunglies
2026-10-06 14:32:59 +08:00
parent 25e00792de
commit dfc729a170
4 changed files with 15 additions and 106 deletions
+3 -50
View File
@@ -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<Vec<Refresh>>;
}
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<T: Send>(operation: impl Future<Output = Result<T>> + Send) -> Result<(T, Vec<Refresh>)> {
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<T: Send>(operation: impl Future<Output = Result<T>> + Send) -> Result<T> {
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));
+1 -1
View File
@@ -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;
+5 -13
View File
@@ -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::*;
+6 -42
View File
@@ -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<Item = Duration> {
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;