refactor(backend): declarative effects (#8128)

* refactor(core): type notice statuses behind the frozen wire contract

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

* refactor(feat): declarative effect registry for verge and clash patches

* refactor(core): derive refresh announcements from transaction commit

* refactor(core): fold tray, hotkey, autostart updates into effect ensures

* refactor(utils): migrate remaining bounded retry loops

* docs: add src-tauri architecture guardrails

* refactor(core): coalesce burst refresh announcements

* refactor(core): complete bounded service retry migration

* ci(backend): add architecture ban gates

* test(feat): freeze representative effect registry mappings

* docs(backend): record dead wire contract entries

* chore(core): document watcher exit conditions

* fix(backend): preserve applied state across later failures

* fix(config): retain runtime after an applied core replacement

* refactor(backend): flatten refresh publication and retry policies

* docs(backend): focus guardrails on architecture ownership

* refactor(effects): use one execution order

* fix(config): publish retained runtime and reconcile proxy settings
This commit is contained in:
Tunglies
2026-10-07 07:38:55 +00:00
committed by GitHub
parent 78691ea5c8
commit 8328e34208
53 changed files with 1925 additions and 876 deletions
+26
View File
@@ -0,0 +1,26 @@
name: Backend Guardrails
on:
pull_request:
paths:
- "src-tauri/**"
- ".github/workflows/backend-guardrails.yml"
workflow_dispatch:
permissions:
contents: read
jobs:
banned-patterns:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- name: No Handle::refresh calls in feat
run: '! grep -rn "Handle::refresh" src-tauri/src/feat'
- name: No notice_message outside core/notify
run: '! grep -rn "notice_message(" src-tauri/src --include="*.rs" | grep -v "core/notify"'
- name: No manual update-flag chains
run: '! grep -rnE "determine_update_flags|UpdateFlags" src-tauri/src'
+24
View File
@@ -0,0 +1,24 @@
# Backend Architecture Guardrails
These rules supplement the repository's root `AGENTS.md`.
- **Configuration ownership.** `config` owns source configuration, drafts and
persistence. The effects layer owns patch orchestration. Rollbackable
transactions publish state after persistence and commit; already applied
runtime and independently committed state remain authoritative even when a
later step fails. Preserve that state, its notifications and the error.
- **Effect ownership.** The declarative registry classifies every Verge patch field,
including fields without effects and platform-specific fields. Effects
reconcile toward the requested state with explicit dependencies and ordering.
Configuration rollback does not imply reversal of OS or lifecycle effects.
- **Shared wire contract.** Notifications are typed. Event names, status values
and payload shapes stay compatible with frontend producers and consumers.
- **Synchronization ownership.** Configuration-write serialization precedes
configuration-update ownership, which precedes Core lifecycle ownership.
When both are held, the lifecycle lock precedes the DNS lock. Independent
transactions retain their ownership boundaries; asynchronous work does not
hold synchronous data guards across suspension.
- **Retry ownership.** Bounded retries share backend policy while preserving
attempts, delays and terminal errors. Background monitors own their lifetime
and cancellation separately; waiting for a condition or traversing distinct
candidates is not a retry policy.
+4 -3
View File
@@ -1,4 +1,5 @@
use super::{CmdResult, CommandFailure, WithErrorCode as _, proxy_aware_coded_error, proxy_aware_error};
use crate::core::notify::NoticeStatus;
use crate::feat;
use crate::utils::{dirs, yaml_emitter};
use crate::{
@@ -56,14 +57,14 @@ pub async fn change_clash_core(clash_core: String) -> CmdResult<Option<CommandFa
match CoreManager::global().restart_core().await {
Ok(_) => {
handle::Handle::notice_message("config_core::change_success", clash_core);
handle::Handle::notice(NoticeStatus::ConfigCoreChangeSuccess, clash_core.as_str());
handle::Handle::refresh_clash();
Ok(None)
}
Err(err) => {
let failed = err.context("core changed but failed to restart");
let error_msg: String = format!("{failed:#}").into();
handle::Handle::notice_message("config_core::change_error", error_msg.clone());
handle::Handle::notice(NoticeStatus::ConfigCoreChangeError, error_msg.as_str());
logging!(error, Type::Core, "core changed but failed to restart: {error_msg}");
Ok(Some(proxy_aware_coded_error(&failed, "CORE_CHANGE_FAILED")))
}
@@ -72,7 +73,7 @@ pub async fn change_clash_core(clash_core: String) -> CmdResult<Option<CommandFa
Err(err) => {
let error_msg: String = format!("{err:#}").into();
logging!(error, Type::Core, "failed to change core: {error_msg}");
handle::Handle::notice_message("config_core::change_error", error_msg);
handle::Handle::notice(NoticeStatus::ConfigCoreChangeError, error_msg.as_str());
Ok(Some(proxy_aware_coded_error(&err, "CORE_CHANGE_FAILED")))
}
}
+1 -1
View File
@@ -1,4 +1,4 @@
use crate::module::lightweight;
use crate::core::lightweight;
use super::CmdResult;
+2 -1
View File
@@ -1,6 +1,7 @@
use super::{CmdResult, StringifyErr as _, WithErrorCode as _, coded_error};
use crate::cmd::validate::{ValidationNoticeTarget, handle_validation_notice};
use crate::config::profiles;
use crate::core::notify::NoticeStatus;
use crate::utils::window_manager::WindowManager;
use crate::{
config::{
@@ -282,7 +283,7 @@ async fn handle_update_error<E: std::fmt::Display>(
logging!(warn, Type::Cmd, "更新过程发生错误: {}", e,);
discard_and_restore(current_profile).await?;
let message: String = e.to_string().into();
handle::Handle::notice_message("config_validate::boot_error", message.clone());
handle::Handle::notice(NoticeStatus::ConfigValidateBootError, message.as_str());
Ok(ValidationOutcome::invalid_from_message(message))
}
+1
View File
@@ -139,6 +139,7 @@ pub async fn sync_tray_proxy_selection() -> CmdResult<()> {
}
async fn run_tray_sync_loop() {
// Permanent watcher: keeps syncing until a pass finds no pending sync request.
loop {
match Tray::global().update_menu().await {
Ok(_) => {
+1 -1
View File
@@ -3,11 +3,11 @@ use crate::{
cmd::StringifyErr as _,
cmd::validate::{ValidationNoticeTarget, handle_validation_notice},
config::{Config, IProfiles, PrfItem},
core::auto_backup::{AutoBackupManager, AutoBackupTrigger},
core::{
CoreManager, handle,
validate::{CoreConfigValidator, ValidationOutcome},
},
module::auto_backup::{AutoBackupManager, AutoBackupTrigger},
utils::dirs,
};
use clash_verge_logging::{Type, logging, logging_error};
+17 -16
View File
@@ -1,3 +1,4 @@
use crate::core::notify::NoticeStatus;
use crate::core::{
handle,
validate::{ValidationErrorKind, ValidationOutcome},
@@ -11,27 +12,27 @@ pub enum ValidationNoticeTarget {
Script,
}
const fn notice_key(kind: ValidationErrorKind, target: ValidationNoticeTarget) -> &'static str {
const fn notice_key(kind: ValidationErrorKind, target: ValidationNoticeTarget) -> NoticeStatus {
match kind {
ValidationErrorKind::FileMissing => "config_validate::file_not_found",
ValidationErrorKind::FileMissing => NoticeStatus::ConfigValidateFileNotFound,
ValidationErrorKind::FileRead => match target {
ValidationNoticeTarget::Script => "config_validate::script_error",
_ => "config_validate::yaml_read_error",
ValidationNoticeTarget::Script => NoticeStatus::ConfigValidateScriptError,
_ => NoticeStatus::ConfigValidateYamlReadError,
},
ValidationErrorKind::YamlSyntax => match target {
ValidationNoticeTarget::Merge => "config_validate::merge_syntax_error",
ValidationNoticeTarget::Script => "config_validate::script_error",
ValidationNoticeTarget::Runtime => "config_validate::yaml_syntax_error",
ValidationNoticeTarget::Merge => NoticeStatus::ConfigValidateMergeSyntaxError,
ValidationNoticeTarget::Script => NoticeStatus::ConfigValidateScriptError,
ValidationNoticeTarget::Runtime => NoticeStatus::ConfigValidateYamlSyntaxError,
},
ValidationErrorKind::YamlMapping => match target {
ValidationNoticeTarget::Merge => "config_validate::merge_mapping_error",
ValidationNoticeTarget::Script => "config_validate::script_error",
ValidationNoticeTarget::Runtime => "config_validate::yaml_mapping_error",
ValidationNoticeTarget::Merge => NoticeStatus::ConfigValidateMergeMappingError,
ValidationNoticeTarget::Script => NoticeStatus::ConfigValidateScriptError,
ValidationNoticeTarget::Runtime => NoticeStatus::ConfigValidateYamlMappingError,
},
ValidationErrorKind::ScriptSyntax => "config_validate::script_syntax_error",
ValidationErrorKind::ScriptMissingMain => "config_validate::script_missing_main",
ValidationErrorKind::ProcessTerminated => "config_validate::process_terminated",
ValidationErrorKind::CoreRejected | ValidationErrorKind::Timeout => "config_validate::error",
ValidationErrorKind::ScriptSyntax => NoticeStatus::ConfigValidateScriptSyntaxError,
ValidationErrorKind::ScriptMissingMain => NoticeStatus::ConfigValidateScriptMissingMain,
ValidationErrorKind::ProcessTerminated => NoticeStatus::ConfigValidateProcessTerminated,
ValidationErrorKind::CoreRejected | ValidationErrorKind::Timeout => NoticeStatus::ConfigValidateError,
}
}
@@ -40,12 +41,12 @@ pub fn handle_validation_notice(outcome: &ValidationOutcome, target: ValidationN
ValidationOutcome::Invalid { kind, message } => {
let status = notice_key(*kind, target);
logging!(warn, Type::Config, "{} 验证失败: {}", file_type, message);
handle::Handle::notice_message(status, message.to_owned());
handle::Handle::notice(status, message.as_str());
}
ValidationOutcome::Busy | ValidationOutcome::Skipped { .. } => {
let message = outcome.to_string();
logging!(warn, Type::Config, "{} 验证跳过: {}", file_type, message);
handle::Handle::notice_message("config_validate::error", message);
handle::Handle::notice(NoticeStatus::ConfigValidateError, message.as_str());
}
ValidationOutcome::Valid => {}
}
+17 -10
View File
@@ -1,4 +1,5 @@
use super::{IClashTemp, IProfiles, IVerge, MixedPort};
use crate::core::notify::NoticeStatus;
use crate::{
config::{PrfItem, profiles_append_item_to_safe, runtime::IRuntime},
constants::{files, timing},
@@ -69,6 +70,12 @@ impl Config {
CONFIG_WRITE_LOCK.lock().await
}
pub(crate) fn try_lock_config_write() -> Result<MutexGuard<'static, ()>> {
CONFIG_WRITE_LOCK
.try_lock()
.map_err(|_| anyhow!("configuration update is already running"))
}
pub async fn init_config_before_window() -> Result<()> {
Self::ensure_default_profile_items().await?;
@@ -106,7 +113,7 @@ impl Config {
if let Some((msg_type, msg_content)) = validation_result {
sleep(timing::STARTUP_ERROR_DELAY).await;
handle::Handle::notice_message(msg_type, msg_content);
handle::Handle::notice(msg_type, msg_content.as_str());
}
Self::runtime().await.apply();
@@ -137,14 +144,14 @@ impl Config {
Ok(())
}
async fn generate_and_validate() -> Result<Option<(&'static str, String)>> {
async fn generate_and_validate() -> Result<Option<(NoticeStatus, String)>> {
if let Err(err) = Self::generate().await {
let error_msg: String = err.to_string().into();
logging!(error, Type::Config, "生成运行时配置失败: {}", error_msg);
CoreManager::global()
.use_default_config("config_validate::boot_error", &error_msg)
.use_default_config(NoticeStatus::ConfigValidateBootError, &error_msg)
.await?;
return Ok(Some(("config_validate::boot_error", error_msg)));
return Ok(Some((NoticeStatus::ConfigValidateBootError, error_msg)));
}
logging!(debug, Type::Config, "生成运行时配置成功");
@@ -171,25 +178,25 @@ impl Config {
error_msg
);
CoreManager::global()
.use_default_config("config_validate::boot_error", &error_msg)
.use_default_config(NoticeStatus::ConfigValidateBootError, &error_msg)
.await?;
Ok(Some(("config_validate::boot_error", error_msg)))
Ok(Some((NoticeStatus::ConfigValidateBootError, error_msg)))
}
Err(err) => {
logging!(warn, Type::Config, "验证过程执行失败: {err:#}");
CoreManager::global()
.use_default_config("config_validate::process_terminated", "")
.use_default_config(NoticeStatus::ConfigValidateProcessTerminated, "")
.await?;
Ok(Some(("config_validate::process_terminated", String::new())))
Ok(Some((NoticeStatus::ConfigValidateProcessTerminated, String::new())))
}
}
} else {
let error_msg = config_result.err().map(|err| err.to_string()).unwrap_or_default();
logging!(warn, Type::Config, "生成配置文件失败,使用默认配置: {error_msg}");
CoreManager::global()
.use_default_config("config_validate::error", "")
.use_default_config(NoticeStatus::ConfigValidateError, "")
.await?;
Ok(Some(("config_validate::error", String::new())))
Ok(Some((NoticeStatus::ConfigValidateError, String::new())))
}
}
+5 -3
View File
@@ -1,3 +1,4 @@
use crate::core::notify::NoticeStatus;
use anyhow::Result;
use clash_verge_draft::DraftTransaction;
use serde::{Deserialize, Serialize};
@@ -138,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_message("dns_override::auto_disabled", "");
Handle::notice(NoticeStatus::DnsOverrideAutoDisabled, "");
}
verge.data_arc().save_file().await
let result = verge.data_arc().save_file().await;
Handle::refresh_verge();
result
}
}
+4 -3
View File
@@ -1,4 +1,5 @@
use super::{Config, IClashTemp, IVerge, MixedPort};
use crate::core::notify::NoticeStatus;
use crate::{
constants::timing,
core::{
@@ -166,8 +167,8 @@ impl Config {
};
AsyncHandler::spawn(move || async move {
tokio::time::sleep(timing::STARTUP_ERROR_DELAY).await;
Handle::notice_message(
"mixed_port::fallback",
Handle::notice(
NoticeStatus::MixedPortFallback,
format!("{},{}", change.original, change.current),
);
});
@@ -196,7 +197,7 @@ fn report_fallback_error(message: String) {
);
AsyncHandler::spawn(move || async move {
tokio::time::sleep(timing::STARTUP_ERROR_DELAY).await;
Handle::notice_message("mixed_port::fallback_error", message);
Handle::notice(NoticeStatus::MixedPortFallbackError, message);
});
}
+155 -148
View File
@@ -2,6 +2,8 @@ use super::{
PrfOption,
prfitem::{PrfItem, PrfSelected, normalize_profile_home_url},
};
use crate::core::notify::{Refresh, announce};
use crate::utils::retry::{RetryError, RetryPolicy, retry, retry_with_state};
use crate::{
core::{handle, tray::Tray},
utils::{
@@ -16,6 +18,7 @@ use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use serde_yaml_ng::Mapping;
use smartstring::alias::String;
use std::num::NonZeroUsize;
use std::{
collections::{HashMap, HashSet},
path::{Component, Path},
@@ -600,19 +603,21 @@ fn is_activation_current(generation: u64) -> bool {
}
async fn fetch_proxies_with_timeout() -> Result<Proxies> {
tokio::time::timeout(MIHOMO_OPERATION_TIMEOUT, async {
loop {
match handle::Handle::mihomo().get_proxies().await {
Ok(proxies) => return proxies,
Err(err) => {
logging!(debug, Type::Config, "mihomo proxies are not ready yet: {err:#}");
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
}
})
let interval = Duration::from_millis(500);
let attempts =
NonZeroUsize::MIN.saturating_add((MIHOMO_OPERATION_TIMEOUT.as_millis() / interval.as_millis()) as usize + 1);
tokio::time::timeout(
MIHOMO_OPERATION_TIMEOUT,
retry(RetryPolicy::fixed(attempts, interval), |_| async {
handle::Handle::mihomo().get_proxies().await.map_err(|error| {
logging!(debug, Type::Config, "mihomo proxies are not ready yet: {error:#}");
RetryError::Retry(error)
})
}),
)
.await
.context("timed out while waiting for mihomo proxies")
.context("timed out while waiting for mihomo proxies")?
.map_err(Into::into)
}
async fn select_node_with_timeout(group_name: &String, node: &String) -> Result<()> {
@@ -691,51 +696,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 +757,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 +791,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.
@@ -836,48 +833,58 @@ fn unsettled_selections(selected: &[PrfSelected], proxies: &Proxies) -> Vec<Stri
/// Retries selections while provider-backed groups finish loading.
async fn settle_pending_selections(selected: &[PrfSelected], completed: &mut HashMap<String, String>, generation: u64) {
let deadline = Instant::now() + SELECTED_NODES_SETTLE_DEADLINE;
loop {
tokio::time::sleep(SELECTED_NODES_SETTLE_INTERVAL).await;
if !is_activation_current(generation) {
return;
}
let Ok(snapshot) = fetch_proxies_with_timeout().await else {
// Unreachable core: the deadline still applies, so this cannot spin forever.
if Instant::now() >= deadline {
return;
}
continue;
};
if !is_activation_current(generation) {
return;
}
let pending = unsettled_selections(selected, &snapshot);
if pending.is_empty() {
return;
}
if Instant::now() >= deadline {
logging!(
warn,
Type::Config,
"gave up putting back {} selected node(s) the core never loaded: {}",
pending.len(),
pending.iter().map(String::as_str).collect::<Vec<_>>().join(", ")
);
return;
}
let plan = reconcile_selected_nodes(selected, None, &snapshot);
if apply_activations(&plan.activations, completed, generation)
.await
.is_none()
{
return;
}
if is_activation_current(generation) {
handle::Handle::refresh_clash();
}
}
let attempts = NonZeroUsize::MIN.saturating_add(
(SELECTED_NODES_SETTLE_DEADLINE.as_millis() / SELECTED_NODES_SETTLE_INTERVAL.as_millis()) as usize,
);
tokio::time::sleep(SELECTED_NODES_SETTLE_INTERVAL).await;
let _ = retry_with_state(
RetryPolicy::fixed(attempts, SELECTED_NODES_SETTLE_INTERVAL),
(selected, completed, generation, deadline),
|state, _| {
Box::pin(async move {
let (selected, completed, generation, deadline) = state;
if !is_activation_current(*generation) {
return Ok(());
}
let Ok(snapshot) = fetch_proxies_with_timeout().await else {
return if Instant::now() >= *deadline {
Ok(())
} else {
Err(RetryError::Retry(()))
};
};
if !is_activation_current(*generation) {
return Ok(());
}
let pending = unsettled_selections(selected, &snapshot);
if pending.is_empty() {
return Ok(());
}
if Instant::now() >= *deadline {
logging!(
warn,
Type::Config,
"gave up putting back {} selected node(s) the core never loaded: {}",
pending.len(),
pending.iter().map(String::as_str).collect::<Vec<_>>().join(", ")
);
return Ok(());
}
let plan = reconcile_selected_nodes(selected, None, &snapshot);
if apply_activations(&plan.activations, completed, *generation)
.await
.is_none()
{
return Ok(());
}
if is_activation_current(*generation) {
announce(Refresh::Clash);
}
Err(RetryError::Retry(()))
})
},
)
.await;
}
/// Releases the first-pass waiter even when restoration exits early.
@@ -920,7 +927,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 +951,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 +1041,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 +1053,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!");
+1 -1
View File
@@ -3,7 +3,7 @@ use anyhow::{Context as _, Result, bail};
use std::{io::ErrorKind, path::PathBuf};
pub(crate) struct FileSnapshot {
path: PathBuf,
pub(crate) path: PathBuf,
content: Option<Vec<u8>>,
}
+4 -3
View File
@@ -1,8 +1,9 @@
use crate::core::notify::NoticeStatus;
use crate::process::AsyncHandler;
use crate::singleton;
use crate::utils::notification::{NotificationEvent, notify_event};
use crate::utils::window_manager::WindowManager;
use crate::{config::Config, core::handle, feat, module::lightweight::entry_lightweight_mode};
use crate::{config::Config, core::handle, core::lightweight::entry_lightweight_mode, feat};
use anyhow::{Result, bail};
use arc_swap::ArcSwap;
use clash_verge_logging::{Type, logging};
@@ -177,7 +178,7 @@ impl Hotkey {
"Hotkey profile reactivation failed validation: {}",
message.as_str()
);
handle::Handle::notice_message("reactivate_profiles::error", message);
handle::Handle::notice(NoticeStatus::ReactivateProfilesError, message);
}
Err(err) => {
logging!(
@@ -186,7 +187,7 @@ impl Hotkey {
"Failed to reactivate subscriptions via hotkey: {}",
err
);
handle::Handle::notice_message("reactivate_profiles::error", err.to_string());
handle::Handle::notice(NoticeStatus::ReactivateProfilesError, err.to_string());
}
});
}
+93 -5
View File
@@ -1,4 +1,5 @@
use super::{CoreManager, PROFILE_SELECTIONS_PENDING_COMMIT, RunningMode};
use crate::core::notify::NoticeStatus;
use crate::core::service::StageRequest;
use crate::{
config::{Config, IProfiles, runtime::IRuntime},
@@ -18,6 +19,7 @@ use smartstring::alias::String;
use std::{
collections::HashSet,
path::PathBuf,
sync::Arc,
time::{Duration, Instant},
};
use tauri_plugin_mihomo::Error as MihomoError;
@@ -129,7 +131,31 @@ where
}
impl CoreManager {
pub async fn use_default_config(&self, error_key: &str, error_msg: &str) -> Result<()> {
pub(crate) fn claim_config_update(
&self,
_config_write: &tokio::sync::MutexGuard<'_, ()>,
) -> Result<ConfigUpdateGuard<'_>> {
if !self.try_start_config_update() {
anyhow::bail!("configuration update is already running");
}
Ok(ConfigUpdateGuard(self))
}
pub(crate) async fn update_config_in_patch(&self, _update: &ConfigUpdateGuard<'_>) -> Result<()> {
if handle::Handle::global().is_exiting() {
anyhow::bail!("application is exiting");
}
self.set_last_update(Instant::now());
Config::generate().await?;
let outcome = self.validate_and_apply_draft().await?;
if outcome.is_valid() {
Ok(())
} else {
Err(anyhow!("{outcome}"))
}
}
pub async fn use_default_config(&self, status: NoticeStatus, message: &str) -> Result<()> {
use crate::constants::files::RUNTIME_CONFIG;
let runtime_path = dirs::app_home_dir()?.join(RUNTIME_CONFIG);
@@ -145,7 +171,7 @@ impl CoreManager {
});
help::save_yaml(&runtime_path, &clash_config, Some("# Clash Verge Runtime")).await?;
handle::Handle::notice_message(error_key, error_msg);
handle::Handle::notice(status, message);
Ok(())
}
@@ -234,6 +260,7 @@ impl CoreManager {
let outcome = self.perform_config_update(Some(profiles)).await?;
if outcome.is_valid() {
crate::config::profiles::restore_selected_nodes().await;
handle::Handle::refresh_clash();
Ok(())
} else {
Err(anyhow!("failed to restore previous Core configuration: {outcome}"))
@@ -297,8 +324,27 @@ impl CoreManager {
self.validate_and_apply(transaction).await
}
/// Validates and applies the caller's transaction, committing only on success.
/// Commits the applied runtime even if restoring the system proxy subsequently fails.
async fn validate_and_apply(&self, transaction: DraftTransaction<'_>) -> Result<ValidationOutcome> {
let runtime = Config::runtime().await;
let original_runtime = runtime.data_arc();
let outcome = match self.validate_and_apply_draft().await {
Ok(outcome) => outcome,
Err(error) => {
transaction.rollback();
if !Arc::ptr_eq(&original_runtime, &runtime.data_arc()) {
handle::Handle::refresh_clash();
}
return Err(error);
}
};
if outcome.is_valid() {
transaction.commit();
}
Ok(outcome)
}
async fn validate_and_apply_draft(&self) -> Result<ValidationOutcome> {
// One serialization feeds check and run files; the core never applies unvalidated bytes.
let yaml = Config::runtime_config_yaml().await?;
let outcome = CoreConfigValidator::global()
@@ -309,8 +355,10 @@ impl CoreManager {
}
let run_path = Config::write_runtime_file(&yaml).await?;
self.apply_config(run_path).await?;
transaction.commit();
let previous = self.current_core_readiness_generation();
let result = self.apply_config(run_path).await;
self.retain_applied_runtime(&Config::runtime().await, previous, &result);
result?;
// Under the lifecycle lock, so a stop for exit waits for this write instead of outrunning it.
#[cfg(target_os = "macos")]
{
@@ -320,6 +368,21 @@ impl CoreManager {
Ok(ValidationOutcome::Valid)
}
fn retain_applied_runtime(
&self,
runtime: &clash_verge_draft::Draft<IRuntime>,
previous: Option<u64>,
result: &Result<()>,
) {
if result.is_ok()
|| self
.current_core_readiness_generation()
.is_some_and(|current| Some(current) != previous)
{
runtime.apply();
}
}
/// Applies a generated configuration through the active core owner.
async fn apply_config(&self, path: PathBuf) -> Result<()> {
if matches!(*self.get_running_mode(), RunningMode::Service) {
@@ -411,6 +474,31 @@ mod tests {
use clash_verge_service_ipc::{StageRejection, StageRuntimeOutcome};
use std::{cell::Cell, time::Duration};
#[test]
fn a_ready_replacement_keeps_the_applied_runtime_when_proxy_restore_fails() -> anyhow::Result<()> {
use crate::{config::runtime::IRuntime, core::CoreManager};
use clash_verge_draft::{Draft, DraftTransaction};
for replaced in [false, true] {
let manager = CoreManager::default();
manager.mark_core_ready();
let previous = manager.current_core_readiness_generation();
let runtime = Draft::new(IRuntime::default());
let transaction = DraftTransaction::begin(vec![&runtime])?;
runtime.edit_draft(|draft| {
draft.exists_keys.insert("replacement".into());
});
if replaced {
manager.mark_core_ready();
}
let result = Err(anyhow::anyhow!("system proxy restore failed"));
manager.retain_applied_runtime(&runtime, previous, &result);
transaction.rollback();
assert_eq!(runtime.data_arc().exists_keys.contains("replacement"), replaced);
}
Ok(())
}
const CONFIRM_WITHIN: Duration = Duration::from_secs(5);
fn staged() -> StageAttempt {
+13 -8
View File
@@ -884,29 +884,34 @@ impl CoreManager {
AsyncHandler::spawn(|| async move {
let manager = Self::global();
let started = Instant::now();
loop {
use crate::utils::retry::{RetryError, RetryPolicy, retry};
let attempts = std::num::NonZeroUsize::MIN.saturating_add(
(timing::SERVICE_HANDOFF_WINDOW.as_millis() / timing::SERVICE_HANDOFF_INTERVAL.as_millis()) as usize,
);
let _ = retry(RetryPolicy::fixed(attempts, std::time::Duration::ZERO), |_| async {
if started.elapsed() >= timing::SERVICE_HANDOFF_WINDOW {
logging!(
info,
Type::Core,
"service handoff window elapsed; staying in sidecar mode"
);
break;
return Ok(());
}
// The window is checked before waiting, including the final handoff attempt.
tokio::time::sleep(timing::SERVICE_HANDOFF_INTERVAL).await;
if !matches!(*manager.get_running_mode(), RunningMode::Sidecar) {
break;
return Ok(());
}
match manager.try_handoff_sidecar_to_service().await {
HandoffOutcome::Done => break,
HandoffOutcome::Done => Ok(()),
HandoffOutcome::Failed => {
logging!(warn, Type::Core, "handoff attempt failed; staying in sidecar mode");
break;
Ok(())
}
HandoffOutcome::NotReady => {}
HandoffOutcome::NotReady => Err(RetryError::Retry(())),
}
}
})
.await;
manager.handoff_watcher_running.store(false, Ordering::Release);
});
}
+45 -36
View File
@@ -1,4 +1,5 @@
mod config;
pub(crate) use config::ConfigUpdateGuard;
mod lifecycle;
mod state;
@@ -237,7 +238,7 @@ impl CoreManager {
}
}
fn current_core_readiness_generation(&self) -> Option<u64> {
pub(crate) fn current_core_readiness_generation(&self) -> Option<u64> {
active_core_readiness_generation(self.core_readiness_state.load(Ordering::Acquire))
}
@@ -284,48 +285,56 @@ impl CoreManager {
anyhow::bail!("core startup blocked after mixed proxy port fallback failure: {reason}");
}
let mut retries = 0;
loop {
match self.start_core().await {
Ok(()) => {
crate::config::Config::notify_startup_mixed_port_fallback();
return Ok(!matches!(*self.get_running_mode(), RunningMode::NotRunning));
}
Err(start_error) if retries < MAX_PORT_FALLBACK_RETRIES => {
if !matches!(*self.get_running_mode(), RunningMode::NotRunning) {
use crate::utils::retry::{RetryError, RetryPolicy, retry};
retry(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add(MAX_PORT_FALLBACK_RETRIES),
std::time::Duration::ZERO,
),
|retries| async move {
match self.start_core().await {
Ok(()) => {
crate::config::Config::notify_startup_mixed_port_fallback();
return Err(start_error);
Ok(!matches!(*self.get_running_mode(), RunningMode::NotRunning))
}
match crate::config::Config::resolve_startup_mixed_port().await {
Ok(true) => {
retries += 1;
tracing::Span::current().record("retries", retries);
logging!(
warn,
Type::Core,
"Retrying core startup after mixed proxy port fallback ({}/{}): {start_error:#}",
retries,
MAX_PORT_FALLBACK_RETRIES
);
}
Ok(false) => {
Err(start_error) if retries < MAX_PORT_FALLBACK_RETRIES => {
if !matches!(*self.get_running_mode(), RunningMode::NotRunning) {
crate::config::Config::notify_startup_mixed_port_fallback();
return Err(start_error);
return Err(RetryError::Stop(start_error));
}
Err(fallback_error) => {
crate::config::Config::block_startup_core(&fallback_error);
return Err(start_error.context(format!(
"the mixed proxy port fallback did not rescue core startup: {fallback_error:#}"
)));
match crate::config::Config::resolve_startup_mixed_port().await {
Ok(true) => {
let retries = retries + 1;
tracing::Span::current().record("retries", retries);
logging!(
warn,
Type::Core,
"Retrying core startup after mixed proxy port fallback ({}/{}): {start_error:#}",
retries,
MAX_PORT_FALLBACK_RETRIES
);
Err(RetryError::Retry(start_error))
}
Ok(false) => {
crate::config::Config::notify_startup_mixed_port_fallback();
Err(RetryError::Stop(start_error))
}
Err(fallback_error) => {
crate::config::Config::block_startup_core(&fallback_error);
Err(RetryError::Stop(start_error.context(format!(
"the mixed proxy port fallback did not rescue core startup: {fallback_error:#}"
))))
}
}
}
Err(error) => {
crate::config::Config::notify_startup_mixed_port_fallback();
Err(RetryError::Stop(error))
}
}
Err(error) => {
crate::config::Config::notify_startup_mixed_port_fallback();
return Err(error);
}
}
}
},
)
.await
}
}
+30 -31
View File
@@ -1,6 +1,7 @@
#[cfg(test)]
use super::claim_core_readiness_generation;
use super::{CoreManager, PROFILE_SELECTIONS_PENDING_COMMIT, RunningMode};
use crate::utils::retry::{RetryError, RetryPolicy, retry};
use crate::{
AsyncHandler,
config::Config,
@@ -11,6 +12,7 @@ use crate::{
use anyhow::{Context as _, Result};
use clash_verge_logging::Type;
use log::Level;
use std::num::NonZeroUsize;
use std::path::Path;
use tauri_plugin_mihomo::MihomoExt as _;
use tauri_plugin_shell::ShellExt as _;
@@ -41,14 +43,15 @@ async fn retry_service_start<Start, StartFuture>(
mut start: Start,
) -> Result<()>
where
Start: FnMut() -> StartFuture,
StartFuture: std::future::Future<Output = Result<()>>,
Start: FnMut() -> StartFuture + Send,
StartFuture: std::future::Future<Output = Result<()>> + Send,
{
let mut last_error = None;
for attempt in 0..attempts {
match start().await {
Ok(()) => return Ok(()),
Err(error) => {
let attempts = NonZeroUsize::new(attempts).ok_or_else(|| anyhow::anyhow!("service start failed"))?;
retry(RetryPolicy::fixed(attempts, retry_delay), |attempt| {
// The Windows Service IPC future would otherwise inflate every enclosing lifecycle future.
let future = Box::pin(start());
async move {
future.await.map_err(|error| {
logging!(
warn,
Type::Core,
@@ -60,16 +63,14 @@ where
.downcast_ref::<service::ServiceStartRefusal>()
.is_some_and(|refusal| service::StageRequest::is_about_the_bundle(refusal.code))
{
return Err(error);
RetryError::Stop(error)
} else {
RetryError::Retry(error)
}
last_error = Some(error);
if attempt + 1 < attempts {
tokio::time::sleep(retry_delay).await;
}
}
})
}
}
Err(last_error.unwrap_or_else(|| anyhow::anyhow!("service start failed")))
})
.await
}
impl CoreManager {
@@ -95,14 +96,16 @@ async fn poll_sidecar_readiness<F, Fut>(
mut probe: F,
) -> Result<()>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<()>>,
F: FnMut() -> Fut + Send,
Fut: std::future::Future<Output = Result<()>> + Send,
{
let mut last_error = None;
for attempt in 0..max_attempts {
match probe().await {
Ok(()) => return Ok(()),
Err(error) => {
let attempts = NonZeroUsize::new(max_attempts)
.context("sidecar readiness was configured with no attempts")
.context("Mihomo API did not become ready")?;
retry(RetryPolicy::fixed(attempts, retry_delay), |attempt| {
let future = probe();
async move {
future.await.map_err(|error| {
logging!(
debug,
Type::Core,
@@ -110,16 +113,12 @@ where
attempt + 1,
max_attempts
);
last_error = Some(error);
}
RetryError::Retry(error)
})
}
if attempt + 1 < max_attempts {
tokio::time::sleep(retry_delay).await;
}
}
Err(last_error
.unwrap_or_else(|| anyhow::anyhow!("sidecar readiness was configured with no attempts"))
.context("Mihomo API did not become ready"))
})
.await
.context("Mihomo API did not become ready")
}
fn should_clear_terminated_sidecar(running_mode: &RunningMode, current_pid: Option<u32>, terminated_pid: u32) -> bool {
+4 -2
View File
@@ -1,13 +1,14 @@
pub mod auto_backup;
pub mod autostart;
pub mod backup;
pub mod handle;
pub mod hotkey;
pub mod lightweight;
pub mod listener;
pub mod logger;
pub mod manager;
#[cfg(target_os = "macos")]
pub mod network_watch;
pub mod notification;
pub mod notify;
pub(crate) mod owner_identity;
pub mod proxy_control;
pub mod proxy_view;
@@ -22,3 +23,4 @@ pub mod validate;
pub mod win_uwp;
pub use self::{manager::CoreManager, timer::Timer, updater::SilentUpdater};
pub use notify::{handle, notification};
+92
View File
@@ -0,0 +1,92 @@
use super::{handle::Handle, notification::FrontendEvent};
use parking_lot::Mutex;
use std::time::Duration;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Refresh {
Clash,
Verge,
Profiles,
Proxies,
}
const REFRESH_WINDOW: Duration = Duration::from_millis(20);
static BATCH: RefreshBatch = RefreshBatch(Mutex::new(Vec::new()));
#[derive(Default)]
struct RefreshBatch(Mutex<Vec<Refresh>>);
impl RefreshBatch {
fn enqueue(&self, refresh: Refresh) -> bool {
let mut pending = self.0.lock();
let start_window = pending.is_empty();
if !pending.contains(&refresh) {
pending.push(refresh);
}
start_window
}
async fn flush(&self, emit: impl Fn(Refresh)) {
// A fixed window bounds latency even under a continuous stream of refreshes.
tokio::time::sleep(REFRESH_WINDOW).await;
let refreshes = std::mem::take(&mut *self.0.lock());
for refresh in refreshes {
emit(refresh);
}
}
}
pub fn announce(refresh: Refresh) {
if BATCH.enqueue(refresh) {
crate::AsyncHandler::spawn(|| BATCH.flush(emit));
}
}
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,
});
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
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));
for refresh in [Refresh::Verge, Refresh::Clash, Refresh::Profiles, Refresh::Proxies] {
assert!(!batch.enqueue(refresh));
}
tokio::time::timeout(Duration::from_secs(1), async {
batch
.flush(|refresh| {
delivered.lock().push(refresh);
if refresh == Refresh::Clash {
assert!(batch.enqueue(Refresh::Clash));
assert!(!batch.enqueue(Refresh::Clash));
}
})
.await;
batch.flush(|refresh| delivered.lock().push(refresh)).await;
})
.await?;
assert_eq!(
*delivered.lock(),
[
Refresh::Clash,
Refresh::Verge,
Refresh::Profiles,
Refresh::Proxies,
Refresh::Clash
]
);
assert!(batch.enqueue(Refresh::Verge));
Ok(())
}
}
@@ -1,6 +1,10 @@
use super::NoticeStatus;
use crate::{APP_HANDLE, singleton};
use smartstring::alias::String;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{
Arc,
atomic::{AtomicBool, Ordering},
};
use tauri::AppHandle;
use tauri_plugin_mihomo::{Mihomo, MihomoExt as _};
@@ -36,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.
@@ -80,13 +76,10 @@ impl Handle {
Self::send_event(FrontendEvent::ProfileUpdateCompleted { uid });
}
pub fn notice_message<S: AsRef<str>, M: Into<String>>(status: S, msg: M) {
let status_str = status.as_ref();
let msg_str = msg.into();
pub fn notice(status: NoticeStatus, message: impl Into<Arc<str>>) {
Self::send_event(FrontendEvent::NoticeMessage {
status: status_str,
message: msg_str,
status,
message: message.into(),
});
}
@@ -102,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;
+208
View File
@@ -0,0 +1,208 @@
mod announce;
pub use announce::{Refresh, announce};
pub mod desktop;
pub mod handle;
pub mod notification;
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
pub enum NoticeStatus {
// dead send: no frontend handler
#[serde(rename = "info")]
Info,
#[serde(rename = "import_sub_url::ok")]
ImportSubUrlOk,
#[serde(rename = "import_sub_url::error")]
ImportSubUrlError,
#[serde(rename = "set_config::error")]
SetConfigError,
#[serde(rename = "core_start::error")]
CoreStartError,
#[serde(rename = "service_core::repair_required")]
#[cfg_attr(not(target_os = "windows"), allow(dead_code))]
ServiceCoreRepairRequired,
#[serde(rename = "service_core::app_data_not_owned")]
ServiceCoreAppDataNotOwned,
#[serde(rename = "service_core::sidecar_fallback")]
#[cfg_attr(not(target_os = "windows"), allow(dead_code))]
ServiceCoreSidecarFallback,
#[serde(rename = "dns_override::auto_disabled")]
DnsOverrideAutoDisabled,
#[serde(rename = "enhance::discarded_keys")]
EnhanceDiscardedKeys,
#[serde(rename = "tun_mode::auto_disabled")]
TunModeAutoDisabled,
#[serde(rename = "tun_mode::auto_disable_failed")]
TunModeAutoDisableFailed,
#[serde(rename = "app_restart::core_stop_failed")]
AppRestartCoreStopFailed,
#[serde(rename = "app_quit::core_stop_failed")]
AppQuitCoreStopFailed,
#[serde(rename = "update_with_clash_proxy")]
UpdateWithClashProxy,
#[serde(rename = "update_failed_even_with_clash")]
UpdateFailedEvenWithClash,
#[serde(rename = "reactivate_profiles::error")]
ReactivateProfilesError,
#[serde(rename = "update_failed")]
UpdateFailed,
#[serde(rename = "config_validate::boot_error")]
ConfigValidateBootError,
#[serde(rename = "config_validate::error")]
ConfigValidateError,
#[serde(rename = "config_validate::process_terminated")]
ConfigValidateProcessTerminated,
#[serde(rename = "config_validate::script_error")]
ConfigValidateScriptError,
#[serde(rename = "config_validate::script_syntax_error")]
ConfigValidateScriptSyntaxError,
#[serde(rename = "config_validate::script_missing_main")]
ConfigValidateScriptMissingMain,
#[serde(rename = "config_validate::file_not_found")]
ConfigValidateFileNotFound,
#[serde(rename = "config_validate::yaml_syntax_error")]
ConfigValidateYamlSyntaxError,
#[serde(rename = "config_validate::yaml_read_error")]
ConfigValidateYamlReadError,
#[serde(rename = "config_validate::yaml_mapping_error")]
ConfigValidateYamlMappingError,
#[serde(rename = "config_validate::merge_syntax_error")]
ConfigValidateMergeSyntaxError,
#[serde(rename = "config_validate::merge_mapping_error")]
ConfigValidateMergeMappingError,
#[serde(rename = "config_core::change_success")]
ConfigCoreChangeSuccess,
#[serde(rename = "config_core::change_error")]
ConfigCoreChangeError,
#[serde(rename = "mixed_port::fallback")]
MixedPortFallback,
#[serde(rename = "mixed_port::fallback_error")]
MixedPortFallbackError,
// dead listener: frontend handler retained
#[allow(dead_code)]
#[serde(rename = "config_validate::core_change")]
ConfigValidateCoreChange,
// dead listener: frontend handler retained
#[allow(dead_code)]
#[serde(rename = "config_validate::stdout_error")]
ConfigValidateStdoutError,
// dead listener: frontend handler retained
#[allow(dead_code)]
#[serde(rename = "config_validate::yaml_key_error")]
ConfigValidateYamlKeyError,
// dead listener: frontend handler retained
#[allow(dead_code)]
#[serde(rename = "config_validate::yaml_error")]
ConfigValidateYamlError,
// dead listener: frontend handler retained
#[allow(dead_code)]
#[serde(rename = "config_validate::merge_key_error")]
ConfigValidateMergeKeyError,
// dead listener: frontend handler retained
#[allow(dead_code)]
#[serde(rename = "config_validate::merge_error")]
ConfigValidateMergeError,
// dead send: no frontend handler
#[serde(rename = "set_config::ok")]
SetConfigOk,
}
#[cfg(test)]
mod tests {
use super::NoticeStatus;
#[test]
fn notice_status_wire_contract() -> Result<(), serde_json::Error> {
let statuses = [
NoticeStatus::Info,
NoticeStatus::ImportSubUrlOk,
NoticeStatus::ImportSubUrlError,
NoticeStatus::SetConfigError,
NoticeStatus::CoreStartError,
NoticeStatus::ServiceCoreRepairRequired,
NoticeStatus::ServiceCoreAppDataNotOwned,
NoticeStatus::ServiceCoreSidecarFallback,
NoticeStatus::DnsOverrideAutoDisabled,
NoticeStatus::EnhanceDiscardedKeys,
NoticeStatus::TunModeAutoDisabled,
NoticeStatus::TunModeAutoDisableFailed,
NoticeStatus::AppRestartCoreStopFailed,
NoticeStatus::AppQuitCoreStopFailed,
NoticeStatus::UpdateWithClashProxy,
NoticeStatus::UpdateFailedEvenWithClash,
NoticeStatus::ReactivateProfilesError,
NoticeStatus::UpdateFailed,
NoticeStatus::ConfigValidateBootError,
NoticeStatus::ConfigValidateError,
NoticeStatus::ConfigValidateProcessTerminated,
NoticeStatus::ConfigValidateScriptError,
NoticeStatus::ConfigValidateScriptSyntaxError,
NoticeStatus::ConfigValidateScriptMissingMain,
NoticeStatus::ConfigValidateFileNotFound,
NoticeStatus::ConfigValidateYamlSyntaxError,
NoticeStatus::ConfigValidateYamlReadError,
NoticeStatus::ConfigValidateYamlMappingError,
NoticeStatus::ConfigValidateMergeSyntaxError,
NoticeStatus::ConfigValidateMergeMappingError,
NoticeStatus::ConfigCoreChangeSuccess,
NoticeStatus::ConfigCoreChangeError,
NoticeStatus::MixedPortFallback,
NoticeStatus::MixedPortFallbackError,
NoticeStatus::ConfigValidateCoreChange,
NoticeStatus::ConfigValidateStdoutError,
NoticeStatus::ConfigValidateYamlKeyError,
NoticeStatus::ConfigValidateYamlError,
NoticeStatus::ConfigValidateMergeKeyError,
NoticeStatus::ConfigValidateMergeError,
NoticeStatus::SetConfigOk,
];
let expected = [
"info",
"import_sub_url::ok",
"import_sub_url::error",
"set_config::error",
"core_start::error",
"service_core::repair_required",
"service_core::app_data_not_owned",
"service_core::sidecar_fallback",
"dns_override::auto_disabled",
"enhance::discarded_keys",
"tun_mode::auto_disabled",
"tun_mode::auto_disable_failed",
"app_restart::core_stop_failed",
"app_quit::core_stop_failed",
"update_with_clash_proxy",
"update_failed_even_with_clash",
"reactivate_profiles::error",
"update_failed",
"config_validate::boot_error",
"config_validate::error",
"config_validate::process_terminated",
"config_validate::script_error",
"config_validate::script_syntax_error",
"config_validate::script_missing_main",
"config_validate::file_not_found",
"config_validate::yaml_syntax_error",
"config_validate::yaml_read_error",
"config_validate::yaml_mapping_error",
"config_validate::merge_syntax_error",
"config_validate::merge_mapping_error",
"config_core::change_success",
"config_core::change_error",
"mixed_port::fallback",
"mixed_port::fallback_error",
"config_validate::core_change",
"config_validate::stdout_error",
"config_validate::yaml_key_error",
"config_validate::yaml_error",
"config_validate::merge_key_error",
"config_validate::merge_error",
"set_config::ok",
];
let actual: Vec<_> = statuses
.into_iter()
.map(serde_json::to_value)
.collect::<Result<_, _>>()?;
assert_eq!(actual, expected.map(serde_json::Value::from));
Ok(())
}
}
@@ -1,8 +1,10 @@
use super::NoticeStatus;
use clash_verge_logging::{Type, logging};
use once_cell::sync::Lazy;
use parking_lot::Mutex;
use serde_json::json;
use smartstring::alias::String;
use std::sync::Arc;
use std::{
collections::HashMap,
future::Future,
@@ -17,8 +19,8 @@ pub enum FrontendEvent<'a> {
RefreshProfiles,
RefreshProxyConfig,
NoticeMessage {
status: &'a str,
message: String,
status: NoticeStatus,
message: Arc<str>,
},
ProfileChanged {
current_profile_id: &'a String,
@@ -323,9 +325,10 @@ impl NotificationSystem {
FrontendEvent::RefreshVerge => ("verge://refresh-verge-config", Ok(json!("yes"))),
FrontendEvent::RefreshProfiles => ("verge://refresh-profiles", Ok(json!("yes"))),
FrontendEvent::RefreshProxyConfig => ("verge://refresh-proxy-config", Ok(serde_json::Value::Null)),
FrontendEvent::NoticeMessage { status, message } => {
("verge://notice-message", serde_json::to_value((status, message)))
}
FrontendEvent::NoticeMessage { status, message } => (
"verge://notice-message",
serde_json::to_value((status, message.as_ref())),
),
FrontendEvent::ProfileChanged { current_profile_id } => ("profile-changed", Ok(json!(current_profile_id))),
FrontendEvent::TimerUpdated { profile_index } => ("verge://timer-updated", Ok(json!(profile_index))),
FrontendEvent::ProfileUpdateStarted { uid } => ("profile-update-started", Ok(json!({ "uid": uid }))),
@@ -353,3 +356,79 @@ impl NotificationSystem {
}
}
}
#[cfg(test)]
mod wire_tests {
use super::*;
#[test]
fn frontend_event_wire_contract() -> Result<(), serde_json::Error> {
let uid = String::from("profile");
let events = [
(
FrontendEvent::RefreshClash,
"verge://refresh-clash-config",
json!("yes"),
),
(
FrontendEvent::RefreshVerge,
"verge://refresh-verge-config",
json!("yes"),
),
(FrontendEvent::RefreshProfiles, "verge://refresh-profiles", json!("yes")),
(
FrontendEvent::RefreshProxyConfig,
"verge://refresh-proxy-config",
json!(null),
),
(
FrontendEvent::NoticeMessage {
status: NoticeStatus::SetConfigOk,
message: Arc::from("ok"),
},
"verge://notice-message",
json!(["set_config::ok", "ok"]),
),
(
FrontendEvent::TimerUpdated { profile_index: &uid },
"verge://timer-updated",
json!("profile"),
),
(
FrontendEvent::RunStateChanged {
state: json!({"running": true}),
},
"verge://run-state-changed",
json!({"running": true}),
),
(
FrontendEvent::PendingFailuresChanged,
"verge://pending-failures-changed",
json!(null),
),
(
FrontendEvent::ProfileChanged {
current_profile_id: &uid,
},
"profile-changed",
json!("profile"),
),
(
FrontendEvent::ProfileUpdateStarted { uid: &uid },
"profile-update-started",
json!({"uid": "profile"}),
),
(
FrontendEvent::ProfileUpdateCompleted { uid: &uid },
"profile-update-completed",
json!({"uid": "profile"}),
),
];
for (event, name, payload) in events {
let (actual_name, actual_payload) = NotificationSystem::serialize_event(event);
assert_eq!(actual_name, name);
assert_eq!(actual_payload?, payload);
}
Ok(())
}
}
+25 -16
View File
@@ -522,25 +522,34 @@ pub async fn clear() -> Result<()> {
/// A failed system call is usually a transient RPC hiccup, so give it a few tries.
async fn clear_with_retry() -> Result<()> {
for attempt in 1..CLEAR_ATTEMPTS {
match clear_inner().await {
Err(error)
use crate::utils::retry::{RetryError, RetryPolicy, retry};
retry(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add((CLEAR_ATTEMPTS - 1) as usize),
CLEAR_RETRY_DELAY,
),
|attempt| async move {
clear_inner().await.map_err(|error| {
if matches!(
SysproxyFailure::from_chain(&error),
Some(SysproxyFailure::SystemCallFailed)
) =>
{
logging!(
warn,
Type::Core,
"clearing the system proxy failed (attempt {attempt}/{CLEAR_ATTEMPTS}); retrying: {error:#}"
);
tokio::time::sleep(CLEAR_RETRY_DELAY).await;
}
other => return other,
}
}
clear_inner().await
) {
if attempt + 1 < CLEAR_ATTEMPTS as usize {
logging!(
warn,
Type::Core,
"clearing the system proxy failed (attempt {}/{CLEAR_ATTEMPTS}); retrying: {error:#}",
attempt + 1
);
}
RetryError::Retry(error)
} else {
RetryError::Stop(error)
}
})
},
)
.await
}
#[tracing::instrument(skip_all, level = "info", fields(route = tracing::field::Empty))]
+13 -12
View File
@@ -130,20 +130,21 @@ impl<E: RunStateEnv> RunStateStore<E> {
/// Retries transport failures but stops at the first readable reply, including a rejection.
pub async fn await_ready(&self, attempts: usize, interval: Duration) -> Result<RunState, ReadyWaitError> {
let mut last_error = None;
for attempt in 0..attempts {
use crate::utils::retry::{RetryError, RetryPolicy, retry};
let attempts = std::num::NonZeroUsize::new(attempts).ok_or_else(|| {
ReadyWaitError::Unreachable(anyhow::anyhow!(
"service readiness wait was configured with no attempts"
))
})?;
retry(RetryPolicy::fixed(attempts, interval), |_| async {
match self.env.probe_service_version().await {
Ok(reply) => return self.record_reply(&reply).map_err(ReadyWaitError::Rejected),
Err(error) => last_error = Some(error),
Ok(reply) => self
.record_reply(&reply)
.map_err(|error| RetryError::Stop(ReadyWaitError::Rejected(error))),
Err(error) => Err(RetryError::Retry(ReadyWaitError::Unreachable(error))),
}
if attempt + 1 < attempts {
tokio::time::sleep(interval).await;
}
}
Err(ReadyWaitError::Unreachable(last_error.unwrap_or_else(|| {
anyhow::anyhow!("service readiness wait was configured with no attempts")
})))
})
.await
}
fn record_reply(&self, reply: &ServiceVersionReply) -> Result<RunState> {
+114 -85
View File
@@ -1,3 +1,4 @@
use crate::core::notify::NoticeStatus;
use crate::utils::dirs;
use crate::{
config::{Config, runtime::IRuntime},
@@ -63,7 +64,7 @@ pub(crate) fn is_not_auto_started(reason: &str) -> bool {
#[cfg(target_os = "windows")]
pub(crate) fn notify_service_fallback(reason: &str) {
*PENDING_SERVICE_FALLBACK_NOTICE.lock() = Some(reason.to_owned());
Handle::notice_message("service_core::sidecar_fallback", "");
Handle::notice(NoticeStatus::ServiceCoreSidecarFallback, "");
}
pub(crate) fn take_service_fallback_notice() -> Option<ServiceFallbackNotice> {
@@ -351,35 +352,40 @@ static SERVICE_CORE_STAGING_GENERATION: AtomicU64 = AtomicU64::new(0);
#[cfg(any(all(target_os = "macos", feature = "verge-dev"), test))]
fn create_service_core_staging_file(directory: &Path, core_name: &std::ffi::OsStr) -> Result<(PathBuf, std::fs::File)> {
for _ in 0..32 {
let generation = SERVICE_CORE_STAGING_GENERATION.fetch_add(1, Ordering::Relaxed);
let temporary_name = format!(
".{}.{}.{generation}.tmp",
core_name.to_string_lossy(),
std::process::id()
);
let temporary_path = directory.join(temporary_name);
match std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temporary_path)
{
Ok(file) => return Ok((temporary_path, file)),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => {
return Err(error).with_context(|| {
use crate::utils::retry::{RetryError, RetryPolicy, retry_sync};
retry_sync(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add(31),
std::time::Duration::ZERO,
),
|_| {
let generation = SERVICE_CORE_STAGING_GENERATION.fetch_add(1, Ordering::Relaxed);
let temporary_name = format!(
".{}.{}.{generation}.tmp",
core_name.to_string_lossy(),
std::process::id()
);
let temporary_path = directory.join(temporary_name);
match std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temporary_path)
{
Ok(file) => Ok((temporary_path, file)),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
Err(RetryError::Retry(anyhow::anyhow!(
"failed to create a unique temporary development Service core in {}",
directory.display()
)))
}
Err(error) => Err(RetryError::Stop(anyhow::Error::from(error).context({
format!(
"failed to create temporary development Service core {}",
temporary_path.display()
)
});
}))),
}
}
}
bail!(
"failed to create a unique temporary development Service core in {}",
directory.display()
},
)
}
@@ -923,11 +929,11 @@ pub(super) async fn start_with_existing_service(config_file: &Path) -> Result<()
#[cfg(target_os = "windows")]
if response.code == ServiceErrorCode::InvalidInstallLocation as u16 {
PENDING_SERVICE_REPAIR_NOTICE.store(true, Ordering::Relaxed);
Handle::notice_message("service_core::repair_required", "");
Handle::notice(NoticeStatus::ServiceCoreRepairRequired, "");
}
if response.code == ServiceErrorCode::AppDataRootNotOwned as u16 {
*PENDING_SERVICE_OWNER_NOTICE.lock() = app_data_owner_command(&credentials);
Handle::notice_message("service_core::app_data_not_owned", "");
Handle::notice(NoticeStatus::ServiceCoreAppDataNotOwned, "");
}
start_owner_monitor();
return Err(record_service_start_refusal(
@@ -1050,38 +1056,47 @@ pub(crate) fn request_runtime_provider_sync(delay: Duration) {
}
AsyncHandler::spawn(move || async move {
tokio::time::sleep(delay).await;
for attempt in 1..=RUNTIME_PROVIDER_SYNC_ATTEMPTS {
let outcome = {
let _serial = PROVIDER_SYNC_SERIAL.lock().await;
if attempt == 1 {
PROVIDER_SYNC_QUEUED.store(false, Ordering::Release);
use crate::utils::retry::{RetryError, RetryPolicy, retry};
let _ = retry(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add((RUNTIME_PROVIDER_SYNC_ATTEMPTS - 1) as usize),
constants::timing::RUNTIME_PROVIDER_SYNC_RETRY_DELAY,
),
|index| async move {
let attempt = index as u32 + 1;
let outcome = {
let _serial = PROVIDER_SYNC_SERIAL.lock().await;
if attempt == 1 {
PROVIDER_SYNC_QUEUED.store(false, Ordering::Release);
}
sync_runtime_providers_by_service().await
};
match outcome {
Ok(ProviderSync { pending: 0, .. }) => Ok(()),
Ok(ProviderSync { pending, .. }) if attempt < RUNTIME_PROVIDER_SYNC_ATTEMPTS => {
logging!(
info,
Type::Service,
"{pending} provider caches are not ready yet; retrying"
);
Err(RetryError::Retry(()))
}
Ok(ProviderSync { pending, .. }) => {
logging!(warn, Type::Service, "{pending} provider caches were not synced");
Ok(())
}
Err(error) => {
logging!(
warn,
Type::Service,
"failed to sync provider caches from the service: {error:#}"
);
Ok(())
}
}
sync_runtime_providers_by_service().await
};
match outcome {
Ok(ProviderSync { pending: 0, .. }) => return,
Ok(ProviderSync { pending, .. }) if attempt < RUNTIME_PROVIDER_SYNC_ATTEMPTS => {
logging!(
info,
Type::Service,
"{pending} provider caches are not ready yet; retrying"
);
}
Ok(ProviderSync { pending, .. }) => {
logging!(warn, Type::Service, "{pending} provider caches were not synced");
return;
}
Err(error) => {
logging!(
warn,
Type::Service,
"failed to sync provider caches from the service: {error:#}"
);
return;
}
}
tokio::time::sleep(constants::timing::RUNTIME_PROVIDER_SYNC_RETRY_DELAY).await;
}
},
)
.await;
});
}
@@ -1389,21 +1404,29 @@ async fn create_sync_temp(target: &Path) -> Result<(PathBuf, tokio::fs::File)> {
.file_name()
.map(|name| name.to_string_lossy())
.unwrap_or_default();
for _ in 0..8 {
let sequence = SYNC_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let temp = target.with_file_name(format!(".{name}.sync-{}-{sequence}.tmp", std::process::id()));
match tokio::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temp)
.await
{
Ok(file) => return Ok((temp, file)),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => return Err(error).with_context(|| format!("failed to create {}", temp.display())),
}
}
bail!("no free temporary name beside {}", target.display())
use crate::utils::retry::{RetryError, RetryPolicy, retry};
retry(
RetryPolicy::fixed(std::num::NonZeroUsize::MIN.saturating_add(7), std::time::Duration::ZERO),
|_| async {
let sequence = SYNC_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let temp = target.with_file_name(format!(".{name}.sync-{}-{sequence}.tmp", std::process::id()));
match tokio::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temp)
.await
{
Ok(file) => Ok((temp, file)),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Err(RetryError::Retry(
anyhow::anyhow!("no free temporary name beside {}", target.display()),
)),
Err(error) => Err(RetryError::Stop(
anyhow::anyhow!(error).context(format!("failed to create {}", temp.display())),
)),
}
},
)
.await
}
async fn same_contents(temp: &Path, target: &Path) -> bool {
@@ -1647,7 +1670,7 @@ fn report_service_core_stopped(status: &ServiceStatusSnapshot) {
);
logging!(error, Type::Service, "service core stopped: {detail}");
CoreManager::global().record_startup_error(CoreFailure::ServiceCoreStopped(detail));
Handle::notice_message("core_start::error", "");
Handle::notice(NoticeStatus::CoreStartError, "");
}
/// Samples ownership, treating every unusable reply as unreadable.
@@ -1776,22 +1799,28 @@ async fn recover_after_owner_loss_while_locked(reason: OwnerRecoveryReason) {
}
async fn clear_proxy_after_owner_loss() {
let mut last_error = None;
for attempt in 1..=3 {
match proxy_control::clear().await {
Ok(()) => return,
Err(error) => {
use crate::utils::retry::{RetryError, RetryPolicy, retry};
let result = retry(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add(2),
Duration::from_millis(100),
),
|index| async move {
let attempt = index + 1;
proxy_control::clear().await.map_err(|error| {
logging!(
warn,
Type::Service,
"proxy clear attempt {attempt}/3 after owner loss failed: {error:#}"
);
last_error = Some(error);
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
}
if let Some(error) = last_error {
RetryError::Retry(error)
})
},
)
.await;
if let Err(error) = result {
// The existing recovery path also waits after its final failed clear.
tokio::time::sleep(Duration::from_millis(100)).await;
logging!(
error,
Type::Service,
+2 -2
View File
@@ -1,14 +1,14 @@
use crate::config::{IProfilePreview, IVerge};
use crate::core::lightweight;
use crate::core::tray::menu_def::TrayAction;
use crate::module::lightweight;
use crate::process::AsyncHandler;
use crate::singleton;
use crate::utils::window_manager::WindowManager;
use crate::{
Type, cmd,
config::Config,
core::lightweight::is_in_lightweight_mode,
feat, logging,
module::lightweight::is_in_lightweight_mode,
utils::{dirs::find_target_icons, help},
};
use clash_verge_limiter::{Limiter, SystemClock, SystemLimiter};
+11 -8
View File
@@ -570,12 +570,14 @@ impl SilentUpdater {
// The new webview may not accept evaluation immediately.
std::thread::spawn(move || {
for i in 0..10 {
std::thread::sleep(std::time::Duration::from_millis(100 * (i + 1)));
if window.eval(&js).is_ok() {
return;
}
}
use crate::utils::retry::{RetryError, RetryPolicy, retry_sync};
let _ = retry_sync(
RetryPolicy::fixed(std::num::NonZeroUsize::MIN.saturating_add(9), std::time::Duration::ZERO),
|index| {
std::thread::sleep(std::time::Duration::from_millis(100 * (index as u64 + 1)));
window.eval(&js).map_err(RetryError::Retry)
},
);
});
logging!(info, Type::System, "Update splash window shown");
@@ -632,8 +634,8 @@ impl SilentUpdater {
Type::System,
"Silent updater: breaking change detected in v{version}, notifying frontend"
);
super::handle::Handle::notice_message(
"info",
super::handle::Handle::notice(
super::notify::NoticeStatus::Info,
format!("New version v{version} contains breaking changes. Please update manually."),
);
return Ok(());
@@ -679,6 +681,7 @@ impl SilentUpdater {
}
pub async fn start_background_check(&self, app_handle: tauri::AppHandle) {
// Permanent watcher: daily update cycles for the lifetime of the app.
logging!(info, Type::System, "Silent updater: background task started");
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+2 -1
View File
@@ -1,3 +1,4 @@
use crate::core::notify::NoticeStatus;
mod chain;
pub mod field;
mod merge;
@@ -436,7 +437,7 @@ fn notify_discarded_keys(keys: Vec<String>) {
drop(notice);
if should_notify {
Handle::notice_message("enhance::discarded_keys", "");
Handle::notice(NoticeStatus::EnhanceDiscardedKeys, "");
}
}
+8 -6
View File
@@ -1,3 +1,5 @@
use crate::core::notify::NoticeStatus;
use crate::core::notify::{Refresh, announce};
use crate::{
config::{Config, MixedPort},
core::{CoreManager, handle, tray},
@@ -26,11 +28,11 @@ 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();
handle::Handle::notice_message("set_config::ok", "ok");
announce(Refresh::Clash);
handle::Handle::notice(NoticeStatus::SetConfigOk, "ok");
}
Err(err) => {
handle::Handle::notice_message("set_config::error", format!("{err:#}"));
handle::Handle::notice(NoticeStatus::SetConfigError, format!("{err:#}"));
logging!(error, Type::Core, "restart core failed: {err:#}");
}
}
@@ -66,8 +68,8 @@ pub async fn restart_app() {
crate::utils::resolve::dns::sync_public_dns().await;
}
}
handle::Handle::notice_message(
"app_restart::core_stop_failed",
handle::Handle::notice(
NoticeStatus::AppRestartCoreStopFailed,
cleanup_result.stop_error.unwrap_or_default(),
);
return;
@@ -108,7 +110,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;
}
+19 -298
View File
@@ -1,279 +1,17 @@
use crate::{
config::{Config, IVerge},
core::{CoreManager, SilentUpdater, autostart, handle, hotkey, logger, proxy_control, tray},
module::{auto_backup::AutoBackupManager, lightweight},
};
use crate::config::{Config, IVerge};
use anyhow::Result;
use bitflags::bitflags;
use clash_verge_draft::{DraftTransaction, SharedDraft};
use clash_verge_logging::{Type, logging, logging_error};
use clash_verge_draft::SharedDraft;
use serde_yaml_ng::Mapping;
use tokio::sync::MutexGuard;
/// Patch Clash configuration
pub async fn patch_clash(patch: &Mapping) -> Result<()> {
Config::clash().await.edit_draft(|d| d.patch_config(patch));
let res = {
// 激活订阅
if patch.get("secret").is_some() || patch.get("external-controller").is_some() {
Config::generate().await?;
CoreManager::global().restart_core().await?;
} else if patch.get("allow-lan").is_some() {
CoreManager::global().update_config_checked().await?;
} else {
if patch.get("mode").is_some() {
tray::Tray::global().update_menu_and_icon().await;
}
CoreManager::global().update_config_checked().await?;
}
handle::Handle::refresh_clash();
<Result<()>>::Ok(())
};
match res {
Ok(()) => {
Config::clash().await.apply();
// 分离数据获取和异步调用
let clash_data = Config::clash().await.data_arc();
clash_data.save_config().await?;
Ok(())
}
Err(err) => {
Config::clash().await.discard();
Err(err)
}
}
}
// Define update flags as bitflags for better performance
bitflags! {
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct UpdateFlags: u16 {
const RESTART_CORE = 1 << 0;
const CLASH_CONFIG = 1 << 1;
const VERGE_CONFIG = 1 << 2;
const LAUNCH = 1 << 3;
const SYS_PROXY = 1 << 4;
const SYSTRAY_ICON = 1 << 5;
const HOTKEY = 1 << 6;
const SYSTRAY_MENU = 1 << 7;
const SYSTRAY_TOOLTIP = 1 << 8;
const SYSTRAY_CLICK_BEHAVIOR = 1 << 9;
const LIGHT_WEIGHT = 1 << 10;
const LANGUAGE = 1 << 11;
const LOG_LEVEL = 1 << 12;
const LOG_FILE = 1 << 13;
const GROUP_SYS_TRAY = Self::SYSTRAY_MENU.bits()
| Self::SYSTRAY_TOOLTIP.bits()
| Self::SYSTRAY_ICON.bits();
}
}
fn determine_update_flags(patch: &IVerge) -> UpdateFlags {
let tun_mode = patch.enable_tun_mode;
let auto_launch = patch.enable_auto_launch;
let system_proxy = patch.enable_system_proxy;
let pac = patch.proxy_auto_config;
let pac_content = &patch.pac_file_content;
let proxy_bypass = &patch.system_proxy_bypass;
let language = &patch.language;
let mixed_port = patch.verge_mixed_port;
#[cfg(target_os = "macos")]
let tray_icon = &patch.tray_icon;
#[cfg(not(target_os = "macos"))]
let tray_icon: Option<String> = None;
let common_tray_icon = patch.common_tray_icon;
let sysproxy_tray_icon = patch.sysproxy_tray_icon;
let tun_tray_icon = patch.tun_tray_icon;
#[cfg(not(target_os = "windows"))]
let redir_enabled = patch.verge_redir_enabled;
#[cfg(not(target_os = "windows"))]
let redir_port = patch.verge_redir_port;
#[cfg(target_os = "linux")]
let tproxy_enabled = patch.verge_tproxy_enabled;
#[cfg(target_os = "linux")]
let tproxy_port = patch.verge_tproxy_port;
let socks_enabled = patch.verge_socks_enabled;
let socks_port = patch.verge_socks_port;
let http_enabled = patch.verge_http_enabled;
let http_port = patch.verge_port;
#[cfg(target_os = "macos")]
let enable_tray_speed = patch.enable_tray_speed;
#[cfg(not(target_os = "macos"))]
let enable_tray_speed: Option<bool> = None;
// let enable_tray_icon = patch.enable_tray_icon;
let enable_global_hotkey = patch.enable_global_hotkey;
let tray_event = &patch.tray_event;
let home_cards = patch.home_cards.as_ref();
let enable_auto_light_weight = patch.enable_auto_light_weight_mode;
let enable_external_controller = patch.enable_external_controller;
let tray_proxy_groups_display_mode = &patch.tray_proxy_groups_display_mode;
let tray_inline_outbound_modes = patch.tray_inline_outbound_modes;
let enable_proxy_guard = patch.enable_proxy_guard;
let proxy_guard_duration = patch.proxy_guard_duration;
let log_level = &patch.app_log_level;
let log_max_size = patch.app_log_max_size;
let log_max_count = patch.app_log_max_count;
#[cfg(target_os = "windows")]
let restart_core_needed = socks_enabled.is_some()
|| http_enabled.is_some()
|| socks_port.is_some()
|| http_port.is_some()
|| mixed_port.is_some()
|| enable_external_controller.is_some();
#[cfg(not(target_os = "windows"))]
let mut restart_core_needed = socks_enabled.is_some()
|| http_enabled.is_some()
|| socks_port.is_some()
|| http_port.is_some()
|| mixed_port.is_some()
|| enable_external_controller.is_some();
#[cfg(not(target_os = "windows"))]
{
restart_core_needed |= redir_enabled.is_some() || redir_port.is_some();
}
#[cfg(target_os = "linux")]
{
restart_core_needed |= tproxy_enabled.is_some() || tproxy_port.is_some();
restart_core_needed |= tun_mode == Some(true);
}
let mut update_flags = UpdateFlags::empty();
if restart_core_needed {
update_flags.insert(UpdateFlags::RESTART_CORE);
}
if tun_mode.is_some() {
update_flags.insert(UpdateFlags::CLASH_CONFIG | UpdateFlags::GROUP_SYS_TRAY);
}
if enable_global_hotkey.is_some() || home_cards.is_some() {
update_flags.insert(UpdateFlags::VERGE_CONFIG);
}
if auto_launch.is_some() {
update_flags.insert(UpdateFlags::LAUNCH);
}
if system_proxy.is_some() {
update_flags.insert(UpdateFlags::SYS_PROXY | UpdateFlags::GROUP_SYS_TRAY);
}
if proxy_bypass.is_some()
|| pac_content.is_some()
|| pac.is_some()
|| enable_proxy_guard.is_some()
|| proxy_guard_duration.is_some()
{
update_flags.insert(UpdateFlags::SYS_PROXY);
}
if language.is_some() {
update_flags.insert(UpdateFlags::LANGUAGE | UpdateFlags::SYSTRAY_MENU | UpdateFlags::SYSTRAY_TOOLTIP);
}
if common_tray_icon.is_some()
|| sysproxy_tray_icon.is_some()
|| tun_tray_icon.is_some()
|| tray_icon.is_some()
|| enable_tray_speed.is_some()
{
update_flags.insert(UpdateFlags::SYSTRAY_ICON);
}
if patch.hotkeys.is_some() {
update_flags.insert(UpdateFlags::HOTKEY | UpdateFlags::SYSTRAY_MENU);
}
if tray_event.is_some() {
update_flags.insert(UpdateFlags::SYSTRAY_CLICK_BEHAVIOR);
}
if enable_auto_light_weight.is_some() {
update_flags.insert(UpdateFlags::LIGHT_WEIGHT);
}
if tray_proxy_groups_display_mode.is_some() {
update_flags.insert(UpdateFlags::SYSTRAY_MENU);
}
if log_level.is_some() {
update_flags.insert(UpdateFlags::LOG_LEVEL);
}
if log_max_size.is_some() || log_max_count.is_some() {
update_flags.insert(UpdateFlags::LOG_FILE);
}
if tray_inline_outbound_modes.is_some() {
update_flags.insert(UpdateFlags::SYSTRAY_MENU);
}
update_flags
}
#[allow(clippy::cognitive_complexity)]
async fn process_terminated_flags(update_flags: UpdateFlags, patch: &IVerge) -> Result<()> {
// Process updates based on flags
if update_flags.contains(UpdateFlags::RESTART_CORE) {
Config::generate().await?;
CoreManager::global().restart_core().await?;
}
if update_flags.contains(UpdateFlags::CLASH_CONFIG) {
CoreManager::global().update_config_checked().await?;
handle::Handle::refresh_clash();
}
if update_flags.contains(UpdateFlags::LAUNCH) {
autostart::update_launch().await?;
}
if update_flags.contains(UpdateFlags::LANGUAGE)
&& let Some(language) = &patch.language
{
clash_verge_i18n::set_locale(language.as_str());
}
if update_flags.contains(UpdateFlags::SYS_PROXY) {
let manager = CoreManager::global();
let _lifecycle = manager.lifecycle_lock.lock().await;
// Turning it off only writes OS state, so it must stay available while the Core is down.
if Config::verge()
.await
.latest_arc()
.enable_system_proxy
.unwrap_or_default()
{
manager.apply_proxy_after_start().await?;
} else {
proxy_control::apply().await?;
proxy_control::refresh_guard().await?;
}
}
if update_flags.contains(UpdateFlags::HOTKEY)
&& let Some(hotkeys) = &patch.hotkeys
{
hotkey::Hotkey::global().update(hotkeys.to_owned()).await?;
}
if update_flags.contains(UpdateFlags::SYSTRAY_MENU) {
tray::Tray::global().update_menu().await?;
}
if update_flags.contains(UpdateFlags::SYSTRAY_ICON) {
tray::Tray::global()
.update_icon(&Config::verge().await.latest_arc())
.await?;
#[cfg(target_os = "macos")]
if patch.enable_tray_speed.is_some() {
tray::Tray::global().update_speed_task(patch.enable_tray_speed.unwrap_or(false));
}
}
if update_flags.contains(UpdateFlags::SYSTRAY_TOOLTIP) {
tray::Tray::global().update_tooltip().await?;
}
if update_flags.contains(UpdateFlags::SYSTRAY_CLICK_BEHAVIOR) {
tray::Tray::global().update_click_behavior().await?;
}
if update_flags.contains(UpdateFlags::LIGHT_WEIGHT) {
if patch.enable_auto_light_weight_mode.unwrap_or(false) {
lightweight::enable_auto_light_weight_mode().await;
} else {
lightweight::disable_auto_light_weight_mode();
}
}
if update_flags.contains(UpdateFlags::LOG_LEVEL) {
logger::Logger::global().update_log_level(patch.get_log_level())?;
}
if update_flags.contains(UpdateFlags::LOG_FILE) {
let log_max_size = patch.app_log_max_size.unwrap_or(128);
let log_max_count = patch.app_log_max_count.unwrap_or(8);
logger::update_log_config(log_max_size, log_max_count).await?;
}
Ok(())
let config_write = Config::try_lock_config_write()?;
super::executor::apply(
&config_write,
super::executor::Patch::Clash(patch),
super::effects::clash_effects(patch),
)
.await
}
/// Apply a patch, then reconcile TUN when its setting changes.
@@ -289,7 +27,7 @@ pub async fn patch_verge(patch: &IVerge, not_save_file: bool) -> Result<()> {
/// Apply a patch without post-update reconciliation.
pub(super) async fn apply_verge_patch(patch: &IVerge, not_save_file: bool) -> Result<()> {
let config_write = Config::lock_config_write().await;
let config_write = Config::try_lock_config_write()?;
apply_verge_patch_locked(&config_write, patch, not_save_file).await
}
@@ -300,32 +38,15 @@ pub(super) async fn apply_verge_patch_locked(
patch: &IVerge,
not_save_file: bool,
) -> Result<()> {
let verge = Config::verge().await;
// Hold the claim across side effects so concurrent transactions cannot share this draft.
let transaction = DraftTransaction::begin(vec![&verge])?;
verge.edit_draft(|d| d.patch_config(patch));
let update_flags = determine_update_flags(patch);
logging!(debug, Type::Setup, "Determined update flags: {:?}", update_flags);
// A failed patch rolls back to what the user already had; it never invents a value for them.
process_terminated_flags(update_flags, patch).await?;
transaction.commit();
if patch.auto_check_update == Some(false) {
SilentUpdater::global().discard_pending();
}
announce_verge_change();
logging_error!(Type::Backup, AutoBackupManager::global().refresh_settings().await);
if !not_save_file {
// 分离数据获取和异步调用
let verge_data = verge.data_arc();
verge_data.save_file().await?;
}
Ok(())
}
fn announce_verge_change() {
handle::Handle::refresh_verge();
super::executor::apply(
_config_write,
super::executor::Patch::Verge {
patch,
persist: !not_save_file,
},
super::effects::verge_effects(patch),
)
.await
}
pub async fn fetch_verge_config() -> Result<SharedDraft<IVerge>> {
+23 -16
View File
@@ -359,24 +359,31 @@ fn stage_core(target: &Path, package: &[u8], version: &str) -> Result<StagedCore
}
fn create_staging_file(directory: &Path, core_name: &OsStr) -> Result<(PathBuf, File)> {
for _ in 0..32 {
let generation = STAGING_GENERATION.fetch_add(1, Ordering::Relaxed);
let path = directory.join(format!(
".{}.{}.{generation}.tmp",
core_name.to_string_lossy(),
std::process::id()
));
use crate::utils::retry::{RetryError, RetryPolicy, retry_sync};
retry_sync(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add(31),
std::time::Duration::ZERO,
),
|_| {
let generation = STAGING_GENERATION.fetch_add(1, Ordering::Relaxed);
let path = directory.join(format!(
".{}.{}.{generation}.tmp",
core_name.to_string_lossy(),
std::process::id()
));
match File::options().write(true).create_new(true).open(&path) {
Ok(file) => return Ok((path, file)),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => {
return Err(error).with_context(|| format!("failed to create {}", path.display()));
match File::options().write(true).create_new(true).open(&path) {
Ok(file) => Ok((path, file)),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Err(RetryError::Retry(
anyhow::anyhow!("failed to create a unique staging file in {}", directory.display()),
)),
Err(error) => Err(RetryError::Stop(
anyhow::Error::from(error).context(format!("failed to create {}", path.display())),
)),
}
}
}
bail!("failed to create a unique staging file in {}", directory.display())
},
)
}
#[cfg(windows)]
+7 -4
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,
};
@@ -74,8 +75,10 @@ pub async fn set_dns_override(
}
transaction.commit();
Handle::refresh_verge();
Handle::refresh_clash();
verge.data_arc().save_file().await?;
// Core and memory state remain applied even when persistence fails.
let result = verge.data_arc().save_file().await;
announce(Refresh::Verge);
announce(Refresh::Clash);
result?;
Ok(outcome)
}
+289
View File
@@ -0,0 +1,289 @@
use crate::config::IVerge;
use std::collections::BTreeSet;
/// Variant order defines execution dependency order.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub(super) enum Effect {
RestartCore,
ClashConfig,
Autostart,
Language,
SystemProxy,
Hotkey,
TrayMenu,
TrayIcon,
TrayTooltip,
TrayClick,
Lightweight,
LogLevel,
LogFile,
}
pub(super) type Effects = BTreeSet<Effect>;
macro_rules! select_field {
($effects:ident, $field:ident, NoEffect) => { let _ = $field; };
($effects:ident, $field:ident, [$($effect:ident),*]) => {
if $field.is_some() { $effects.extend([$(Effect::$effect),*]); }
};
($effects:ident, $field:ident, Tun) => {
select_field!($effects, $field, [ClashConfig, TrayMenu, TrayTooltip, TrayIcon]);
#[cfg(target_os = "linux")]
if *$field == Some(true) { $effects.insert(Effect::RestartCore); }
};
}
macro_rules! verge_registry {
($( $(#[$cfg:meta])* $field:ident => $rule:tt; )*) => {
pub(super) fn verge_effects(patch: &IVerge) -> Effects {
// No rest pattern: even fields omitted by serde must be classified at compile time.
let IVerge { $( $(#[$cfg])* $field, )* } = patch;
let mut effects = Effects::new();
$( $(#[$cfg])* { select_field!(effects, $field, $rule); } )*
effects
}
};
}
verge_registry! {
app_log_level => [LogLevel];
app_log_max_size => [LogFile];
app_log_max_count => [LogFile];
language => [Language, TrayMenu, TrayTooltip];
theme_mode => NoEffect;
tray_event => [TrayClick];
env_type => NoEffect;
start_page => NoEffect;
startup_script => NoEffect;
traffic_graph => NoEffect;
enable_memory_usage => NoEffect;
enable_group_icon => NoEffect;
pause_render_traffic_stats_on_blur => NoEffect;
common_tray_icon => [TrayIcon];
#[cfg(target_os = "macos")]
tray_icon => [TrayIcon];
menu_icon => NoEffect;
menu_order => NoEffect;
proxy_group_tools_position => NoEffect;
notice_position => NoEffect;
collapse_navbar => NoEffect;
sysproxy_tray_icon => [TrayIcon];
tun_tray_icon => [TrayIcon];
enable_tun_mode => Tun;
enable_auto_launch => [Autostart];
enable_silent_start => NoEffect;
enable_system_proxy => [SystemProxy, TrayMenu, TrayTooltip, TrayIcon];
enable_proxy_guard => [SystemProxy];
enable_bypass_check => NoEffect;
enable_dns_settings => NoEffect;
profile_dns_settings => NoEffect;
use_default_bypass => [SystemProxy];
system_proxy_bypass => [SystemProxy];
proxy_guard_duration => [SystemProxy];
proxy_auto_config => [SystemProxy];
pac_file_content => [SystemProxy];
proxy_host => [SystemProxy];
theme_setting => NoEffect;
web_ui_list => NoEffect;
clash_core => NoEffect;
hotkeys => [Hotkey, TrayMenu];
enable_global_hotkey => NoEffect;
home_cards => NoEffect;
auto_close_connection => NoEffect;
auto_check_update => NoEffect;
default_latency_test => NoEffect;
default_latency_timeout => NoEffect;
enable_auto_delay_detection => NoEffect;
auto_delay_detection_interval_minutes => NoEffect;
enable_builtin_enhanced => NoEffect;
proxy_layout_column => NoEffect;
test_list => NoEffect;
auto_log_clean => NoEffect;
enable_auto_backup_schedule => NoEffect;
auto_backup_interval_hours => NoEffect;
auto_backup_on_change => NoEffect;
#[cfg(not(target_os = "windows"))]
verge_redir_port => [RestartCore];
#[cfg(not(target_os = "windows"))]
verge_redir_enabled => [RestartCore];
#[cfg(target_os = "linux")]
verge_tproxy_port => [RestartCore];
#[cfg(target_os = "linux")]
verge_tproxy_enabled => [RestartCore];
verge_mixed_port => [RestartCore];
verge_socks_port => [RestartCore];
verge_socks_enabled => [RestartCore];
verge_port => [RestartCore];
verge_http_enabled => [RestartCore];
webdav_url => NoEffect;
webdav_username => NoEffect;
webdav_password => NoEffect;
#[cfg(target_os = "macos")]
enable_tray_speed => [TrayIcon];
tray_proxy_groups_display_mode => [TrayMenu];
tray_inline_outbound_modes => [TrayMenu];
enable_auto_light_weight_mode => [Lightweight];
auto_light_weight_minutes => NoEffect;
enable_hover_jump_navigator => NoEffect;
hover_jump_navigator_delay => NoEffect;
enable_external_controller => [RestartCore];
}
pub(super) fn clash_effects(patch: &serde_yaml_ng::Mapping) -> Effects {
use clash_verge_logging::{Type, logging};
let mut effects = Effects::new();
for key in patch.keys() {
effects.extend(match key.as_str() {
Some("secret" | "external-controller") => &[Effect::RestartCore][..],
Some("allow-lan") => &[Effect::ClashConfig][..],
Some("mode") => &[Effect::ClashConfig, Effect::TrayMenu, Effect::TrayIcon][..],
_ => {
logging!(
warn,
Type::Config,
"Unregistered Clash configuration key {key:?}; reloading configuration"
);
&[Effect::ClashConfig][..]
}
});
}
// These branches were exclusive: restart supersedes reload, and allow-lan precedes mode.
if effects.contains(&Effect::RestartCore) {
effects.retain(|effect| *effect == Effect::RestartCore);
} else if patch.contains_key("allow-lan") {
effects.retain(|effect| *effect == Effect::ClashConfig);
}
if patch.is_empty() {
effects.insert(Effect::ClashConfig);
}
effects
}
#[cfg(test)]
mod tests {
use super::*;
fn verge_patch(mutate: impl FnOnce(&mut IVerge)) -> IVerge {
let mut patch = IVerge::default();
mutate(&mut patch);
patch
}
#[test]
fn representative_verge_patches_map_to_frozen_effects() {
assert_eq!(verge_effects(&IVerge::default()), Effects::new());
let cases: [(IVerge, &[Effect]); 9] = [
(
verge_patch(|p| p.verge_mixed_port = Some(27899)),
&[Effect::RestartCore],
),
(
verge_patch(|p| p.language = Some("zh".into())),
&[Effect::TrayMenu, Effect::TrayTooltip, Effect::Language],
),
(
verge_patch(|p| p.hotkeys = Some(vec!["CmdOrCtrl+K".into()])),
&[Effect::Hotkey, Effect::TrayMenu],
),
(
verge_patch(|p| p.enable_system_proxy = Some(true)),
&[
Effect::SystemProxy,
Effect::TrayIcon,
Effect::TrayMenu,
Effect::TrayTooltip,
],
),
(
verge_patch(|p| p.system_proxy_bypass = Some("localhost".into())),
&[Effect::SystemProxy],
),
(
verge_patch(|p| p.use_default_bypass = Some(false)),
&[Effect::SystemProxy],
),
(
verge_patch(|p| p.proxy_host = Some("localhost".into())),
&[Effect::SystemProxy],
),
(
verge_patch(|p| p.tray_event = Some("show_main_window".into())),
&[Effect::TrayClick],
),
(verge_patch(|p| p.theme_mode = Some("dark".into())), &[]),
];
for (patch, expected) in cases {
assert_eq!(
verge_effects(&patch),
expected.iter().copied().collect::<Effects>(),
"patch mapping drifted for {expected:?}"
);
}
let tun = verge_patch(|p| p.enable_tun_mode = Some(true));
#[cfg(target_os = "linux")]
assert_eq!(
verge_effects(&tun),
[
Effect::RestartCore,
Effect::ClashConfig,
Effect::TrayIcon,
Effect::TrayMenu,
Effect::TrayTooltip,
]
.into_iter()
.collect::<Effects>()
);
#[cfg(not(target_os = "linux"))]
assert_eq!(
verge_effects(&tun),
[
Effect::ClashConfig,
Effect::TrayIcon,
Effect::TrayMenu,
Effect::TrayTooltip,
]
.into_iter()
.collect::<Effects>()
);
}
fn clash_mapping(pairs: &[(&str, &str)]) -> serde_yaml_ng::Mapping {
let mut mapping = serde_yaml_ng::Mapping::new();
for (key, value) in pairs {
mapping.insert((*key).into(), (*value).into());
}
mapping
}
#[test]
fn clash_keys_map_to_frozen_effects() {
let cases: [(serde_yaml_ng::Mapping, &[Effect]); 7] = [
(clash_mapping(&[]), &[Effect::ClashConfig]),
(clash_mapping(&[("secret", "s")]), &[Effect::RestartCore]),
(
clash_mapping(&[("mode", "rule")]),
&[Effect::ClashConfig, Effect::TrayIcon, Effect::TrayMenu],
),
(clash_mapping(&[("allow-lan", "true")]), &[Effect::ClashConfig]),
(clash_mapping(&[("not-a-known-key", "1")]), &[Effect::ClashConfig]),
// Restart supersedes reload; allow-lan precedes mode.
(
clash_mapping(&[("secret", "s"), ("mode", "rule")]),
&[Effect::RestartCore],
),
(
clash_mapping(&[("allow-lan", "true"), ("mode", "rule")]),
&[Effect::ClashConfig],
),
];
for (patch, expected) in cases {
assert_eq!(
clash_effects(&patch),
expected.iter().copied().collect::<Effects>(),
"clash mapping drifted for {expected:?}"
);
}
}
}
+196
View File
@@ -0,0 +1,196 @@
use super::effects::{Effect, Effects};
use crate::core::notify::{Refresh, announce};
use crate::{
config::{
Config, IVerge,
snapshot::{capture_config_files, restore_files},
},
core::auto_backup::AutoBackupManager,
core::{
CoreManager, SilentUpdater, autostart, hotkey, lightweight, logger, manager::ConfigUpdateGuard, proxy_control,
tray,
},
};
use anyhow::Result;
use clash_verge_draft::DraftTransaction;
use clash_verge_logging::{Type, logging_error};
use serde_yaml_ng::Mapping;
use std::sync::Arc;
use tokio::sync::MutexGuard;
pub(super) enum Patch<'a> {
Verge { patch: &'a IVerge, persist: bool },
Clash(&'a Mapping),
}
async fn ensure_restart(manager: &CoreManager) -> Result<()> {
Config::generate().await?;
let previous = manager.current_core_readiness_generation();
let result = manager.restart_core_during_config_update().await;
if result.is_ok()
|| manager
.current_core_readiness_generation()
.is_some_and(|current| Some(current) != previous)
{
Config::runtime().await.apply();
}
result?;
Ok(())
}
async fn ensure_effect(
effect: Effect,
patch: &IVerge,
manager: &CoreManager,
update: &ConfigUpdateGuard<'_>,
) -> Result<()> {
match effect {
Effect::RestartCore => {
ensure_restart(manager).await?;
}
Effect::ClashConfig => {
manager.update_config_in_patch(update).await?;
}
Effect::Autostart => autostart::update_launch().await?,
Effect::Language => {
if let Some(language) = &patch.language {
clash_verge_i18n::set_locale(language.as_str());
}
}
Effect::SystemProxy => {
let _lifecycle = manager.lifecycle_lock.lock().await;
// Disabling only writes OS state and must remain available when the Core is stopped.
if Config::verge()
.await
.latest_arc()
.enable_system_proxy
.unwrap_or_default()
{
manager.apply_proxy_after_start().await?;
} else {
proxy_control::apply().await?;
proxy_control::refresh_guard().await?;
}
}
Effect::Hotkey => {
if let Some(hotkeys) = &patch.hotkeys {
hotkey::Hotkey::global().update(hotkeys.to_owned()).await?;
}
}
Effect::TrayMenu => tray::Tray::global().update_menu().await?,
Effect::TrayIcon => {
tray::Tray::global()
.update_icon(&Config::verge().await.latest_arc())
.await?;
#[cfg(target_os = "macos")]
if let Some(enabled) = patch.enable_tray_speed {
tray::Tray::global().update_speed_task(enabled);
}
}
Effect::TrayTooltip => tray::Tray::global().update_tooltip().await?,
Effect::TrayClick => tray::Tray::global().update_click_behavior().await?,
Effect::Lightweight => {
if patch.enable_auto_light_weight_mode.unwrap_or(false) {
lightweight::enable_auto_light_weight_mode().await;
} else {
lightweight::disable_auto_light_weight_mode();
}
}
Effect::LogLevel => logger::Logger::global().update_log_level(patch.get_log_level())?,
Effect::LogFile => {
logger::update_log_config(
patch.app_log_max_size.unwrap_or(128),
patch.app_log_max_count.unwrap_or(8),
)
.await?;
}
}
Ok(())
}
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;
let verge = Config::verge().await;
let runtime = Config::runtime().await;
let transaction = match patch {
Patch::Verge { .. } => DraftTransaction::begin(vec![&verge, &runtime])?,
Patch::Clash(_) => DraftTransaction::begin(vec![&clash, &runtime])?,
};
let original_runtime = runtime.data_arc();
let mut snapshots = capture_config_files().await?;
let empty = IVerge::default();
let verge_patch = match &patch {
Patch::Verge { patch, .. } => patch,
Patch::Clash(_) => &empty,
};
match &patch {
Patch::Verge { patch, .. } => verge.edit_draft(|draft| draft.patch_config(patch)),
Patch::Clash(patch) => clash.edit_draft(|draft| draft.patch_config(patch)),
}
let result: Result<()> = async {
for effect in effects.iter().copied() {
let result = Box::pin(ensure_effect(effect, verge_patch, manager, &update)).await;
if matches!(patch, Patch::Clash(_)) && matches!(effect, Effect::TrayMenu | Effect::TrayIcon) {
// Tray failures must not reject an applied Clash mode change.
logging_error!(Type::Tray, result);
} else {
result?;
}
}
match patch {
Patch::Verge { persist: true, .. } => verge.latest_arc().save_file().await?,
Patch::Verge { persist: false, .. } => {}
Patch::Clash(_) => clash.latest_arc().save_config().await?,
}
Ok(())
}
.await;
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);
}
return match restore_files(&snapshots).await {
Ok(()) => Err(error),
Err(rollback_error) => Err(proxy_control::rollback_failure(error, rollback_error)),
};
}
transaction.commit();
match patch {
Patch::Verge { patch, .. } => {
if patch.auto_check_update == Some(false) {
SilentUpdater::global().discard_pending();
}
if effects.contains(&Effect::ClashConfig) {
announce(Refresh::Clash);
}
logging_error!(Type::Backup, AutoBackupManager::global().refresh_settings().await);
announce(Refresh::Verge);
}
Patch::Clash(_) => announce(Refresh::Clash),
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn config_claim_is_exclusive_and_released_on_drop() -> Result<()> {
let write_lock = tokio::sync::Mutex::new(());
let write = write_lock.lock().await;
let manager = CoreManager::default();
let first = manager.claim_config_update(&write)?;
assert!(manager.claim_config_update(&write).is_err());
drop(first);
assert!(manager.claim_config_update(&write).is_ok());
drop(write);
Ok(())
}
}
+3 -3
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,
@@ -126,8 +126,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)
}
+2
View File
@@ -3,6 +3,8 @@ mod clash;
mod config;
mod core_upgrade;
mod dns;
mod effects;
mod executor;
mod icon;
mod listener;
mod profile;
+74 -56
View File
@@ -1,3 +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},
@@ -43,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;
}
@@ -100,7 +102,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,
@@ -108,7 +110,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
@@ -117,66 +119,82 @@ 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_message("update_with_clash_proxy", profile_name);
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_message("update_with_clash_proxy", profile_name);
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());
if is_mannual_trigger {
handle::Handle::notice_message("update_failed_even_with_clash", format!("{profile_name} - {last_err}"));
handle::Handle::notice(
NoticeStatus::UpdateFailedEvenWithClash,
format!("{profile_name} - {last_err}"),
);
}
bail!(last_err)
}
@@ -201,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);
@@ -218,7 +236,7 @@ pub async fn update_profile(uid: &String, option: Option<&PrfOption>, is_mannual
message
};
logging!(error, Type::Config, "[订阅更新] 更新失败: {}", message);
handle::Handle::notice_message("update_failed", &message);
handle::Handle::notice(NoticeStatus::UpdateFailed, message.as_str());
bail!(message);
}
}
+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)
}
+4 -2
View File
@@ -1,5 +1,7 @@
//! Reconciles TUN with current Core and Service availability.
use crate::core::notify::NoticeStatus;
use std::sync::atomic::{AtomicBool, Ordering};
use clash_verge_logging::{Type, logging};
@@ -75,10 +77,10 @@ pub async fn reconcile_tun_availability() {
// Skip recursive reconciliation for this internal patch.
match super::apply_verge_patch_locked(&config_write, &patch, false).await {
Ok(()) => Handle::notice_message("tun_mode::auto_disabled", ""),
Ok(()) => Handle::notice(NoticeStatus::TunModeAutoDisabled, ""),
Err(error) => {
logging!(error, Type::Core, "failed to turn TUN mode off: {error:#}");
Handle::notice_message("tun_mode::auto_disable_failed", "");
Handle::notice(NoticeStatus::TunModeAutoDisableFailed, "");
}
}
}
+5 -4
View File
@@ -1,6 +1,7 @@
use crate::config::Config;
use crate::core::lightweight;
use crate::core::notify::NoticeStatus;
use crate::core::{CoreManager, handle};
use crate::module::lightweight;
use crate::utils;
use crate::utils::window_manager::WindowManager;
use anyhow::{Context as _, anyhow};
@@ -160,8 +161,8 @@ pub async fn quit() -> clash_verge_signal::ShutdownOutcome {
crate::utils::resolve::dns::sync_public_dns().await;
}
}
handle::Handle::notice_message(
"app_quit::core_stop_failed",
handle::Handle::notice(
NoticeStatus::AppQuitCoreStopFailed,
cleanup_result.stop_error.unwrap_or_default(),
);
return clash_verge_signal::ShutdownOutcome::Canceled;
@@ -300,7 +301,7 @@ async fn clean_session_ending_best_effort() -> CleanupResult {
#[cfg(target_os = "macos")]
pub async fn hide() {
use crate::module::lightweight::add_light_weight_timer;
use crate::core::lightweight::add_light_weight_timer;
let enable_auto_light_weight_mode = Config::verge()
.await
+2 -3
View File
@@ -7,7 +7,6 @@ mod constants;
mod core;
mod enhance;
mod feat;
mod module;
mod process;
pub mod utils;
@@ -336,7 +335,7 @@ pub fn run() -> std::process::ExitCode {
mod event_handlers {
#[cfg(target_os = "macos")]
use crate::module::lightweight;
use crate::core::lightweight;
use crate::utils::window_manager::WindowManager;
use crate::{
config::Config,
@@ -492,7 +491,7 @@ pub fn run() -> std::process::ExitCode {
}
#[allow(unused_variables)]
tauri::RunEvent::ExitRequested { api, code, .. } => {
if module::lightweight::is_in_lightweight_mode() && !handle::Handle::global().is_exiting() {
if core::lightweight::is_in_lightweight_mode() && !handle::Handle::global().is_exiting() {
api.prevent_exit();
} else if code.is_none() {
api.prevent_exit();
-2
View File
@@ -1,2 +0,0 @@
pub mod auto_backup;
pub mod lightweight;
+2 -1
View File
@@ -1,3 +1,4 @@
use crate::core::notify::NoticeStatus;
use crate::{config::with_encryption, process::AsyncHandler};
use anyhow::{Context as _, Result, anyhow, bail};
use clash_verge_logging::{Type, logging};
@@ -58,7 +59,7 @@ fn parse_mapping(yaml_str: String, path: PathBuf) -> Result<Mapping> {
let error_msg = format!("YAML syntax error in {}: {}", path.display(), err);
logging!(error, Type::Config, "{}", error_msg);
crate::core::handle::Handle::notice_message("config_validate::yaml_syntax_error", &error_msg);
crate::core::handle::Handle::notice(NoticeStatus::ConfigValidateYamlSyntaxError, error_msg.as_str());
bail!("YAML syntax error: {}", err)
}
+53 -36
View File
@@ -109,44 +109,61 @@ fn current_user_home() -> anyhow::Result<PathBuf> {
}
.max(1024);
loop {
if buffer_size > MAX_BUFFER_SIZE {
anyhow::bail!("effective-user home lookup exceeded the maximum buffer size");
}
use crate::utils::retry::{RetryError, RetryPolicy, retry_sync};
retry_sync(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add(11),
std::time::Duration::ZERO,
),
|_| {
if buffer_size > MAX_BUFFER_SIZE {
return Err(RetryError::Stop(anyhow::anyhow!(
"effective-user home lookup exceeded the maximum buffer size"
)));
}
let mut passwd = std::mem::MaybeUninit::<libc::passwd>::zeroed();
let mut result = std::ptr::null_mut();
let mut buffer = vec![0_u8; buffer_size];
let code = unsafe {
libc::getpwuid_r(
uid,
passwd.as_mut_ptr(),
buffer.as_mut_ptr().cast(),
buffer.len(),
&mut result,
)
};
if code == libc::ERANGE {
buffer_size = buffer_size.saturating_mul(2);
continue;
}
if code != 0 {
return Err(std::io::Error::from_raw_os_error(code).into());
}
if result.is_null() {
anyhow::bail!("effective user {uid} has no password database entry");
}
let mut passwd = std::mem::MaybeUninit::<libc::passwd>::zeroed();
let mut result = std::ptr::null_mut();
let mut buffer = vec![0_u8; buffer_size];
let code = unsafe {
libc::getpwuid_r(
uid,
passwd.as_mut_ptr(),
buffer.as_mut_ptr().cast(),
buffer.len(),
&mut result,
)
};
if code == libc::ERANGE {
buffer_size = buffer_size.saturating_mul(2);
return Err(RetryError::Retry(anyhow::anyhow!(
"effective-user home lookup exceeded the maximum buffer size"
)));
}
if code != 0 {
return Err(RetryError::Stop(std::io::Error::from_raw_os_error(code).into()));
}
if result.is_null() {
return Err(RetryError::Stop(anyhow::anyhow!(
"effective user {uid} has no password database entry"
)));
}
let passwd = unsafe { passwd.assume_init() };
if passwd.pw_dir.is_null() {
anyhow::bail!("effective user {uid} has no home directory");
}
let home = unsafe { CStr::from_ptr(passwd.pw_dir) };
if home.to_bytes().is_empty() {
anyhow::bail!("effective user {uid} has an empty home directory");
}
return Ok(PathBuf::from(std::ffi::OsStr::from_bytes(home.to_bytes())));
}
let passwd = unsafe { passwd.assume_init() };
if passwd.pw_dir.is_null() {
return Err(RetryError::Stop(anyhow::anyhow!(
"effective user {uid} has no home directory"
)));
}
let home = unsafe { CStr::from_ptr(passwd.pw_dir) };
if home.to_bytes().is_empty() {
return Err(RetryError::Stop(anyhow::anyhow!(
"effective user {uid} has an empty home directory"
)));
}
Ok(PathBuf::from(std::ffi::OsStr::from_bytes(home.to_bytes())))
},
)
}
pub const fn move_decision(
+2 -1
View File
@@ -8,9 +8,10 @@ pub mod linux;
#[cfg(target_os = "macos")]
pub mod macos_launch_guard;
pub(crate) mod network;
pub mod notification;
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;
+3 -2
View File
@@ -1,3 +1,4 @@
use crate::core::notify::NoticeStatus;
use std::sync::atomic::{AtomicBool, Ordering};
use anyhow::Result;
@@ -13,8 +14,8 @@ use crate::{
service::{SERVICE_MANAGER, ServiceManager},
tray::Tray,
},
core::{auto_backup::AutoBackupManager, lightweight::auto_lightweight_boot},
feat,
module::{auto_backup::AutoBackupManager, lightweight::auto_lightweight_boot},
process::AsyncHandler,
utils::{init, server, window_manager::WindowManager},
};
@@ -224,7 +225,7 @@ async fn init_core_manager() -> bool {
Err(error) => {
logging!(error, Type::Setup, "core manager initialization failed: {error:#}");
CoreManager::global().record_startup_error(CoreFailure::StartFailed(format!("{error:#}")));
Handle::notice_message("core_start::error", "");
Handle::notice(NoticeStatus::CoreStartError, "");
false
}
}
+8 -7
View File
@@ -1,3 +1,4 @@
use crate::core::notify::NoticeStatus;
use anyhow::Result;
use percent_encoding::percent_decode_str;
use smartstring::alias::String;
@@ -93,18 +94,18 @@ async fn import_subscription(url: &str, name: Option<&String>) {
if let Err(e) = profiles::profiles_append_item_safe(&mut item).await {
logging!(error, Type::Config, "failed to import subscription url: {:?}", e);
Config::profiles().await.discard();
handle::Handle::notice_message("import_sub_url::error", e.to_string());
handle::Handle::notice(NoticeStatus::ImportSubUrlError, e.to_string());
return;
}
if let Err(e) = profiles::profiles_save_file_safe().await {
logging!(error, Type::Config, "failed to save imported subscription: {e:#}");
handle::Handle::notice_message("import_sub_url::error", e.to_string());
handle::Handle::notice(NoticeStatus::ImportSubUrlError, e.to_string());
return;
}
logging_error!(Type::Timer, Timer::global().refresh().await);
handle::Handle::notice_message(
"import_sub_url::ok",
handle::Handle::notice(
NoticeStatus::ImportSubUrlOk,
"", // 空 msg 传入,我们不希望导致 后端-前端-后端 死循环,这里只做提醒。
);
@@ -116,7 +117,7 @@ async fn fetch_profile_item(url: &str, name: Option<&String>) -> Option<PrfItem>
Ok(item) => Some(item),
Err(e) => {
logging!(error, Type::Config, "failed to parse profile from url: {:?}", e);
handle::Handle::notice_message("import_sub_url::error", e.to_string());
handle::Handle::notice(NoticeStatus::ImportSubUrlError, e.to_string());
None
}
}
@@ -149,11 +150,11 @@ async fn refresh_core_config() {
Ok(outcome) => {
let message = outcome.to_string();
logging!(warn, Type::Config, "Apply config failed: {}", message);
handle::Handle::notice_message("config_validate::error", message);
handle::Handle::notice(NoticeStatus::ConfigValidateError, message);
}
Err(err) => {
logging!(error, Type::Config, "Apply config error: {err:#}");
handle::Handle::notice_message("update_failed", format!("{err}"));
handle::Handle::notice(NoticeStatus::UpdateFailed, format!("{err}"));
}
}
}
+185
View File
@@ -0,0 +1,185 @@
use std::{future::Future, num::NonZeroUsize, time::Duration};
#[derive(Clone, Copy, Debug)]
pub struct RetryPolicy {
pub attempts: NonZeroUsize,
delay: Duration,
}
impl RetryPolicy {
pub const fn fixed(attempts: NonZeroUsize, delay: Duration) -> Self {
Self { attempts, delay }
}
fn delays(self) -> impl Iterator<Item = Duration> {
std::iter::repeat_n(self.delay, self.attempts.get() - 1)
}
}
pub enum RetryError<E> {
Retry(E),
Stop(E),
}
pub async fn retry<T: Send, E: Send, Fut: Future<Output = Result<T, RetryError<E>>> + Send>(
policy: RetryPolicy,
mut operation: impl FnMut(usize) -> Fut + Send,
) -> 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 + Send,
T: Send,
E: Send,
Fut: Future<Output = Result<T, E>> + Send,
const N: usize,
>(
first: S,
rest: [S; N],
mut operation: impl FnMut(S) -> Fut + Send,
) -> Result<(S, T), E> {
retry(
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);
async move { future.await.map(|value| (strategy, value)).map_err(RetryError::Retry) }
},
)
.await
}
pub async fn retry_with_state<S: Send, T: Send, E: Send>(
policy: RetryPolicy,
mut state: S,
mut operation: impl for<'a> FnMut(&'a mut S, usize) -> futures::future::BoxFuture<'a, Result<T, RetryError<E>>> + Send,
) -> Result<T, E> {
let mut result = operation(&mut state, 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(&mut state, index + 1).await;
}
result.map_err(|error| match error {
RetryError::Retry(error) | RetryError::Stop(error) => error,
})
}
pub fn retry_sync<T, E>(
policy: RetryPolicy,
mut operation: impl FnMut(usize) -> Result<T, RetryError<E>>,
) -> Result<T, E> {
let mut result = operation(0);
for (index, delay) in policy.delays().enumerate() {
match result {
Ok(_) | Err(RetryError::Stop(_)) => break,
Err(RetryError::Retry(_)) => {
if !delay.is_zero() {
std::thread::sleep(delay);
}
}
}
result = operation(index + 1);
}
result.map_err(|error| match error {
RetryError::Retry(error) | RetryError::Stop(error) => error,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn retries_stop_at_success_terminal_error_or_attempt_limit() {
let policy = RetryPolicy::fixed(NonZeroUsize::MIN.saturating_add(2), Duration::ZERO);
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"]);
}
#[tokio::test]
async fn stateful_and_sync_retries_preserve_progress_and_stop() {
let policy = RetryPolicy::fixed(NonZeroUsize::MIN.saturating_add(2), Duration::ZERO);
let result = retry_with_state(policy, Vec::new(), |visited, attempt| {
Box::pin(async move {
visited.push(attempt);
if visited.len() == 3 {
Ok(visited.clone())
} else {
Err(RetryError::Retry(()))
}
})
})
.await;
assert_eq!(result, Ok(vec![0, 1, 2]));
let mut visited = Vec::new();
let result = retry_sync(policy, |attempt| {
visited.push(attempt);
if attempt == 1 {
Err::<(), _>(RetryError::Stop("terminal"))
} else {
Err(RetryError::Retry("retry"))
}
});
assert_eq!(result, Err("terminal"));
assert_eq!(visited, [0, 1]);
}
}
+29 -15
View File
@@ -1,7 +1,7 @@
use super::resolve;
use crate::{
config::{Config, DEFAULT_PAC, MixedPort},
module::lightweight,
core::lightweight,
process::AsyncHandler,
utils::{dirs, window_manager::WindowManager},
};
@@ -54,20 +54,34 @@ pub async fn check_singleton() -> Result<SingletonDisposition> {
.context("failed to initialize singleton lock")?;
if !try_lock_instance(&lock).context("failed to acquire singleton lock")? {
let deadline = std::time::Instant::now() + Duration::from_secs(20);
loop {
if std::time::Instant::now() >= deadline {
bail!("another app instance is starting");
}
// Retry the lock: during restart, the old server stops before releasing it.
if try_lock_instance(&lock).context("failed to acquire singleton lock")? {
break;
}
if let Ok(record) = read_instance_record(&record_path)
&& notify_existing_instance(&record).await
{
return Ok(SingletonDisposition::Secondary);
}
tokio::time::sleep(Duration::from_millis(50)).await;
use crate::utils::retry::{RetryError, RetryPolicy, retry};
let secondary = retry(
RetryPolicy::fixed(
std::num::NonZeroUsize::MIN.saturating_add(400),
Duration::from_millis(50),
),
|_| async {
if std::time::Instant::now() >= deadline {
return Err(RetryError::Stop(anyhow::anyhow!("another app instance is starting")));
}
// A restarting primary releases the lock after spawning us, with its server down.
if try_lock_instance(&lock)
.context("failed to acquire singleton lock")
.map_err(RetryError::Stop)?
{
return Ok(false);
}
if let Ok(record) = read_instance_record(&record_path)
&& notify_existing_instance(&record).await
{
return Ok(true);
}
Err(RetryError::Retry(anyhow::anyhow!("another app instance is starting")))
},
)
.await?;
if secondary {
return Ok(SingletonDisposition::Secondary);
}
}