refactor(core): derive refresh announcements from transaction commit

This commit is contained in:
Tunglies
2026-10-01 14:34:16 +08:00
committed by GitHub
parent 438f6f7461
commit 7aed7be2f1
11 changed files with 211 additions and 125 deletions
+3 -2
View File
@@ -139,12 +139,13 @@ impl Config {
}
// The runtime is already committed; a disk failure cannot undo its effective settings.
transaction.commit();
Handle::refresh_verge();
if state.requested && !state.enabled {
PENDING_DNS_OVERRIDE_NOTICE.store(true, Ordering::Relaxed);
Handle::notice(NoticeStatus::DnsOverrideAutoDisabled, "");
}
verge.data_arc().save_file().await
verge.data_arc().save_file().await?;
Handle::refresh_verge();
Ok(())
}
}
+88 -95
View File
@@ -2,6 +2,7 @@ use super::{
PrfOption,
prfitem::{PrfItem, PrfSelected, normalize_profile_home_url},
};
use crate::core::notify::{Refresh, announce};
use crate::{
core::{handle, tray::Tray},
utils::{
@@ -691,51 +692,58 @@ async fn update_tray_after_activation(generation: u64) {
}
}
async fn persist_selected_update<F, Fut>(supersede: bool, edit: F) -> Result<()>
where
F: FnOnce(IProfiles) -> Fut + Send,
Fut: std::future::Future<Output = Result<(IProfiles, bool)>> + Send,
{
let changed = Config::profiles().await.with_data_modify(edit).await?;
if changed {
if supersede {
supersede_selected_activation();
}
announce(Refresh::Profiles);
}
Ok(())
}
/// Records a backend-made selection so the next core start restores it.
pub(crate) async fn record_selected_node(group_name: &str, node: &str) -> Result<()> {
let group_name = String::from(group_name);
let node = String::from(node);
let recorded = Config::profiles()
.await
.with_data_modify(move |mut profiles| async move {
let Some(current) = profiles.current.clone() else {
return Ok((profiles, false));
};
let Some(item) = profiles
.items
.as_mut()
.and_then(|items| items.iter_mut().find(|item| item.uid.as_ref() == Some(&current)))
else {
return Ok((profiles, false));
};
persist_selected_update(true, move |mut profiles| async move {
let Some(current) = profiles.current.clone() else {
return Ok((profiles, false));
};
let Some(item) = profiles
.items
.as_mut()
.and_then(|items| items.iter_mut().find(|item| item.uid.as_ref() == Some(&current)))
else {
return Ok((profiles, false));
};
let mut selected = item.selected.clone().unwrap_or_default();
match selected
.iter_mut()
.find(|entry| entry.name.as_ref() == Some(&group_name))
{
Some(entry) => {
if entry.now.as_ref() == Some(&node) {
return Ok((profiles, false));
}
entry.now = Some(node);
let mut selected = item.selected.clone().unwrap_or_default();
match selected
.iter_mut()
.find(|entry| entry.name.as_ref() == Some(&group_name))
{
Some(entry) => {
if entry.now.as_ref() == Some(&node) {
return Ok((profiles, false));
}
None => selected.push(PrfSelected {
name: Some(group_name),
now: Some(node),
}),
entry.now = Some(node);
}
item.selected = Some(selected);
profiles.save_file().await?;
Ok((profiles, true))
})
.await?;
if recorded {
supersede_selected_activation();
handle::Handle::refresh_profiles();
}
Ok(())
None => selected.push(PrfSelected {
name: Some(group_name),
now: Some(node),
}),
}
item.selected = Some(selected);
profiles.save_file().await?;
Ok((profiles, true))
})
.await
}
fn remove_selected_node(selected: &mut Vec<PrfSelected>, group_name: &str) -> bool {
@@ -745,36 +753,28 @@ fn remove_selected_node(selected: &mut Vec<PrfSelected>, group_name: &str) -> bo
}
pub(crate) async fn forget_selected_node(group_name: &str) -> Result<()> {
let cleared = Config::profiles()
.await
.with_data_modify(move |mut profiles| async move {
let Some(current) = profiles.current.clone() else {
return Ok((profiles, false));
};
let Some(item) = profiles
.items
.as_mut()
.and_then(|items| items.iter_mut().find(|item| item.uid.as_ref() == Some(&current)))
else {
return Ok((profiles, false));
};
persist_selected_update(true, move |mut profiles| async move {
let Some(current) = profiles.current.clone() else {
return Ok((profiles, false));
};
let Some(item) = profiles
.items
.as_mut()
.and_then(|items| items.iter_mut().find(|item| item.uid.as_ref() == Some(&current)))
else {
return Ok((profiles, false));
};
let mut selected = item.selected.clone().unwrap_or_default();
if !remove_selected_node(&mut selected, group_name) {
return Ok((profiles, false));
}
let mut selected = item.selected.clone().unwrap_or_default();
if !remove_selected_node(&mut selected, group_name) {
return Ok((profiles, false));
}
item.selected = (!selected.is_empty()).then_some(selected);
profiles.save_file().await?;
Ok((profiles, true))
})
.await?;
if cleared {
supersede_selected_activation();
handle::Handle::refresh_profiles();
}
Ok(())
item.selected = (!selected.is_empty()).then_some(selected);
profiles.save_file().await?;
Ok((profiles, true))
})
.await
}
async fn persist_reconciled_selected(
@@ -787,34 +787,27 @@ async fn persist_reconciled_selected(
return Ok(());
}
let profiles = Config::profiles().await;
let profile_uid = profile_uid.clone();
let original_selected = original_selected.to_vec();
let updated = profiles
.with_data_modify(move |mut profiles| async move {
if !is_activation_current(generation) || profiles.current.as_ref() != Some(&profile_uid) {
return Ok((profiles, false));
}
persist_selected_update(false, move |mut profiles| async move {
if !is_activation_current(generation) || profiles.current.as_ref() != Some(&profile_uid) {
return Ok((profiles, false));
}
let profile = profiles
.items
.as_mut()
.and_then(|items| items.iter_mut().find(|item| item.uid.as_ref() == Some(&profile_uid)))
.with_context(|| format!("failed to find the profile item \"uid:{profile_uid}\""))?;
if profile.selected.as_deref().unwrap_or(&[]) != original_selected.as_slice() {
return Ok((profiles, false));
}
let profile = profiles
.items
.as_mut()
.and_then(|items| items.iter_mut().find(|item| item.uid.as_ref() == Some(&profile_uid)))
.with_context(|| format!("failed to find the profile item \"uid:{profile_uid}\""))?;
if profile.selected.as_deref().unwrap_or(&[]) != original_selected.as_slice() {
return Ok((profiles, false));
}
profile.selected = (!selected.is_empty()).then_some(selected);
profiles.save_file().await?;
Ok((profiles, true))
})
.await?;
if updated {
handle::Handle::refresh_profiles();
}
Ok(())
profile.selected = (!selected.is_empty()).then_some(selected);
profiles.save_file().await?;
Ok((profiles, true))
})
.await
}
/// Returns recorded selections whose groups have not reached the requested node.
@@ -875,7 +868,7 @@ async fn settle_pending_selections(selected: &[PrfSelected], completed: &mut Has
return;
}
if is_activation_current(generation) {
handle::Handle::refresh_clash();
announce(Refresh::Clash);
}
}
}
@@ -920,7 +913,7 @@ async fn activate_selected_nodes_worker(
}
if is_activation_current(generation) {
handle::Handle::refresh_clash();
announce(Refresh::Clash);
}
let plan = if needs_confirmation {
@@ -944,7 +937,7 @@ async fn activate_selected_nodes_worker(
return Ok(());
};
if confirmed_activated_count > 0 && is_activation_current(generation) {
handle::Handle::refresh_clash();
announce(Refresh::Clash);
}
confirmed_plan
} else {
@@ -1034,7 +1027,7 @@ fn activate_selected_nodes_with(repair: SelectionRepair) -> tokio::sync::oneshot
if selected.is_empty() {
if is_activation_current(generation) {
handle::Handle::refresh_clash();
announce(Refresh::Clash);
}
return Ok(());
}
@@ -1046,7 +1039,7 @@ fn activate_selected_nodes_with(repair: SelectionRepair) -> tokio::sync::oneshot
if let Err(err) = result {
logging!(error, Type::Config, "failed to activate selected nodes: {err:#}");
// The profile itself is already active even if node restoration failed.
handle::Handle::refresh_clash();
announce(Refresh::Clash);
}
update_tray_after_activation(generation).await;
logging!(debug, Type::Config, "activating selected nodes done!");
+77
View File
@@ -0,0 +1,77 @@
use super::{handle::Handle, notification::FrontendEvent};
use anyhow::Result;
use std::{cell::RefCell, future::Future};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Refresh {
Clash,
Verge,
Profiles,
Proxies,
}
tokio::task_local! {
static PENDING: RefCell<Vec<Refresh>>;
}
pub fn announce(refresh: Refresh) {
if PENDING.try_with(|pending| pending.borrow_mut().push(refresh)).is_err() {
emit(refresh);
}
}
fn emit(refresh: Refresh) {
Handle::send_event(match refresh {
Refresh::Clash => FrontendEvent::RefreshClash,
Refresh::Verge => FrontendEvent::RefreshVerge,
Refresh::Profiles => FrontendEvent::RefreshProfiles,
Refresh::Proxies => FrontendEvent::RefreshProxyConfig,
});
}
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(())
}
}
+3 -11
View File
@@ -40,19 +40,11 @@ impl Handle {
}
pub fn refresh_clash() {
Self::send_event(FrontendEvent::RefreshClash);
super::announce(super::Refresh::Clash);
}
pub fn refresh_verge() {
Self::send_event(FrontendEvent::RefreshVerge);
}
pub fn refresh_profiles() {
Self::send_event(FrontendEvent::RefreshProfiles);
}
pub fn refresh_proxy_config() {
Self::send_event(FrontendEvent::RefreshProxyConfig);
super::announce(super::Refresh::Verge);
}
/// Push a Run State snapshot to the frontend.
@@ -103,7 +95,7 @@ impl Handle {
self.is_exiting.load(Ordering::Acquire)
}
fn send_event(event: FrontendEvent) {
pub(super) fn send_event(event: FrontendEvent) {
let handle = Self::global();
if handle.is_exiting() {
return;
+2
View File
@@ -1,3 +1,5 @@
mod announce;
pub use announce::{Refresh, after_commit, announce};
pub mod desktop;
pub mod handle;
pub mod notification;
+3 -2
View File
@@ -1,4 +1,5 @@
use crate::core::notify::NoticeStatus;
use crate::core::notify::{Refresh, announce};
use crate::{
config::{Config, MixedPort},
core::{CoreManager, handle, tray},
@@ -27,7 +28,7 @@ static TLS_CONFIG: Lazy<Arc<rustls::ClientConfig>> = Lazy::new(|| {
pub async fn restart_clash_core() {
match CoreManager::global().restart_core().await {
Ok(_) => {
handle::Handle::refresh_clash();
announce(Refresh::Clash);
handle::Handle::notice(NoticeStatus::SetConfigOk, "ok");
}
Err(err) => {
@@ -96,7 +97,7 @@ pub async fn change_clash_mode(mode: String) -> Result<(), String> {
let clash_data = clash.data_arc();
if clash_data.save_config().await.is_ok() {
handle::Handle::refresh_clash();
announce(Refresh::Clash);
tray::Tray::global().update_menu_and_icon().await;
}
+12 -3
View File
@@ -1,3 +1,4 @@
use crate::core::notify::{Refresh, announce};
use anyhow::{Result, ensure};
use clash_verge_draft::DraftTransaction;
use serde::Serialize;
@@ -10,7 +11,7 @@ use crate::{
profiles::PROFILE_WRITE_LOCK,
},
constants::files::DNS_CONFIG,
core::{CoreManager, handle::Handle},
core::CoreManager,
utils::dirs,
};
@@ -25,6 +26,14 @@ pub async fn set_dns_override(
profile_uid: String,
enabled: bool,
confirmation: Option<String>,
) -> Result<DnsOverrideOutcome> {
crate::core::notify::after_commit(Box::pin(set_dns_override_inner(profile_uid, enabled, confirmation))).await
}
async fn set_dns_override_inner(
profile_uid: String,
enabled: bool,
confirmation: Option<String>,
) -> Result<DnsOverrideOutcome> {
let _profile_write = PROFILE_WRITE_LOCK.lock().await;
let _config_write = Config::lock_config_write().await;
@@ -74,8 +83,8 @@ pub async fn set_dns_override(
}
transaction.commit();
Handle::refresh_verge();
Handle::refresh_clash();
verge.data_arc().save_file().await?;
announce(Refresh::Verge);
announce(Refresh::Clash);
Ok(outcome)
}
+10 -5
View File
@@ -1,10 +1,11 @@
use super::effects::{Effect, Effects};
use crate::core::notify::{Refresh, announce};
use crate::{
config::{
Config, IVerge,
snapshot::{capture_config_files, restore_files},
},
core::{CoreManager, autostart, handle::Handle, hotkey, logger, manager::ConfigUpdateGuard, proxy_control, tray},
core::{CoreManager, autostart, hotkey, logger, manager::ConfigUpdateGuard, proxy_control, tray},
module::{auto_backup::AutoBackupManager, lightweight},
};
use anyhow::Result;
@@ -58,7 +59,7 @@ ensure!(ensure_restart, ctx, {
});
ensure!(ensure_clash, ctx, {
ctx.manager.update_config_in_patch(ctx.update).await?;
Handle::refresh_clash();
announce(Refresh::Clash);
Ok(())
});
ensure!(ensure_verge, _ctx, { Ok(()) });
@@ -128,7 +129,7 @@ ensure!(ensure_log_file, ctx, {
.await
});
pub(super) async fn apply(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> {
async fn apply_inner(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;
@@ -177,13 +178,17 @@ pub(super) async fn apply(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, e
match patch {
Patch::Verge { .. } => {
logging_error!(Type::Backup, AutoBackupManager::global().refresh_settings().await);
Handle::refresh_verge();
announce(Refresh::Verge);
}
Patch::Clash(_) => Handle::refresh_clash(),
Patch::Clash(_) => announce(Refresh::Clash),
}
Ok(())
}
pub(super) async fn apply(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> {
crate::core::notify::after_commit(apply_inner(config_write, patch, effects)).await
}
#[cfg(test)]
mod tests {
use super::*;
+8 -4
View File
@@ -1,3 +1,4 @@
use crate::core::notify::{Refresh, announce};
use crate::{
config::{
Config, MixedPort,
@@ -5,7 +6,6 @@ use crate::{
},
core::{
CoreManager,
handle::Handle,
listener::{
ListenerProbe, ListenerProbeOutcome, ProxyPortSettings, SaveProxyPortsOutcome,
probe_listener as probe_listener_sync, probe_proxy_port_change,
@@ -27,8 +27,12 @@ pub async fn probe_listener(request: ListenerProbe) -> Result<ListenerProbeOutco
.context("listener probe task failed")
}
#[allow(clippy::cognitive_complexity)]
pub async fn save_proxy_ports(settings: ProxyPortSettings) -> Result<SaveProxyPortsOutcome> {
crate::core::notify::after_commit(Box::pin(save_proxy_ports_inner(settings))).await
}
#[allow(clippy::cognitive_complexity)]
async fn save_proxy_ports_inner(settings: ProxyPortSettings) -> Result<SaveProxyPortsOutcome> {
settings.validate()?;
let _config_write = Config::lock_config_write().await;
let manager = CoreManager::global();
@@ -126,8 +130,8 @@ pub async fn save_proxy_ports(settings: ProxyPortSettings) -> Result<SaveProxyPo
transaction.commit();
// The save landed, so the port the app had borrowed is now irrelevant.
let _ = ScopeGuard::into_inner(borrowed_port);
Handle::refresh_clash();
Handle::refresh_verge();
announce(Refresh::Clash);
announce(Refresh::Verge);
logging!(info, Type::Config, "Proxy port configuration applied and persisted");
Ok(SaveProxyPortsOutcome::Saved)
}
+3 -2
View File
@@ -1,4 +1,5 @@
use crate::core::notify::NoticeStatus;
use crate::core::notify::{Refresh, announce};
use crate::{
cmd,
config::{Config, PrfItem, PrfOption, profiles::profiles_update_item_safe},
@@ -44,7 +45,7 @@ pub async fn switch_proxy_node(group_name: &str, proxy_name: &str) {
{
Ok(_) => {
record_switched_node(group_name, proxy_name).await;
handle::Handle::refresh_proxy_config();
announce(Refresh::Proxies);
let _ = tray::Tray::global().update_menu().await;
return;
}
@@ -218,7 +219,7 @@ pub async fn update_profile(uid: &String, option: Option<&PrfOption>, is_mannual
match CoreManager::global().update_config_with_force(is_mannual_trigger).await {
Ok(outcome) if outcome.is_valid() => {
logging_error!(Type::Config, Config::sync_dns_override().await);
handle::Handle::refresh_clash();
announce(Refresh::Clash);
}
Ok(outcome @ (ValidationOutcome::Skipped { .. } | ValidationOutcome::Busy)) if !is_mannual_trigger => {
logging!(info, Type::Config, "[订阅更新] 本次配置刷新已跳过: {}", outcome);
+2 -1
View File
@@ -1,3 +1,4 @@
use crate::core::notify::{Refresh, announce};
use crate::{
config::{Config, IVerge, MixedPort},
core::{
@@ -83,7 +84,7 @@ pub async fn toggle_tun_mode(not_save_file: Option<bool>) -> bool {
.await
{
Ok(_) => {
handle::Handle::refresh_verge();
announce(Refresh::Verge);
// Reconciliation may immediately disable unavailable TUN; report the resulting state.
Config::verge().await.latest_arc().enable_tun_mode.unwrap_or(false)
}