gitea-admin/clash-verge-rev
1
0
Fork 0
mirror of https://github.com/clash-verge-rev/clash-verge-rev.git synced 2026-11-09 03:42:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: safely restore selected proxy nodes

Browse Source
This commit is contained in:
wonfen
2026-07-16 04:16:42 +08:00
parent 6c149d6234
commit cd20bdb6a1
6 changed files with 471 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/config/prfitem.rs
+1 -1
View File
@@ -63,7 +63,7 @@ pub struct PrfItem {
pub file_data: Option<String>,
}
#[derive(Default, Debug, Clone, Deserialize, Serialize)]
#[derive(Default, Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
pub struct PrfSelected {
pub name: Option<String>,
pub now: Option<String>,
src-tauri/src/config/profiles.rs
+449 -75
View File
@@ -1,4 +1,7 @@
use super::{PrfOption, prfitem::PrfItem};
use super::{
PrfOption,
prfitem::{PrfItem, PrfSelected},
};
use crate::{
core::handle,
utils::{
@@ -12,7 +15,15 @@ use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use serde_yaml_ng::Mapping;
use smartstring::alias::String;
use std::{collections::HashSet, sync::LazyLock, time::Duration};
use std::{
collections::HashSet,
sync::{
LazyLock,
atomic::{AtomicU64, Ordering},
},
time::Duration,
};
use tauri_plugin_mihomo::models::{Proxies, ProxyType};
use tokio::{fs, task::JoinHandle};
/// Regex to check profile file names, eg.
@@ -29,6 +40,12 @@ static REGEX_PROFILE_FILE: LazyLock<regex::Regex> =
// activate selected nodes task handle
static ACTIVATE_SELECTED_TASK: LazyLock<Mutex<Option<JoinHandle<()>>>> = LazyLock::new(|| Mutex::new(None));
static ACTIVATE_SELECTED_GENERATION: AtomicU64 = AtomicU64::new(0);
// The plugin already limits the request/response phase to 5 seconds. This outer timeout also covers
// lock acquisition, connection-pool waiting, and local-socket connection establishment.
const MIHOMO_OPERATION_TIMEOUT: Duration = Duration::from_secs(10);
const SELECTED_NODES_RECHECK_DELAY: Duration = Duration::from_secs(1);
/// Define the `profiles.yaml` schema
#[derive(Default, Debug, Clone, Deserialize, Serialize)]
@@ -592,89 +609,446 @@ pub async fn profiles_draft_update_item_safe(index: &String, item: &mut PrfItem)
.await
}
pub async fn activate_selected_nodes() -> Result<()> {
logging!(info, Type::Config, "starting activating selected nodes");
let value = ACTIVATE_SELECTED_TASK.lock().take();
if let Some(handle) = value {
logging!(info, Type::Config, "aborting previous worker");
handle.abort();
}
let profiles = Config::profiles().await.latest_arc();
let Some(current) = profiles.get_current() else {
bail!("no current profile running");
};
let profile = profiles
.get_item(current)
.context("failed to get current profile")?
.clone();
#[derive(Debug, PartialEq, Eq)]
struct SelectedNodesPlan {
selected: Vec<PrfSelected>,
activations: Vec<(String, String)>,
repaired_count: usize,
}
let handle = tokio::spawn(async move {
let mihomo = handle::Handle::mihomo().await;
// check mihomo is running
let mut is_mihomo_ready = false;
for _ in 0..10 {
if mihomo.get_version().await.is_ok() {
logging!(debug, Type::Config, "check mihomo api success");
is_mihomo_ready = true;
break;
fn node_is_available(available_nodes: &[std::string::String], node: &str) -> bool {
available_nodes.iter().any(|available| available == node)
}
fn selected_nodes_need_confirmation(selected: &[PrfSelected], proxies: &Proxies) -> bool {
selected.iter().any(|selected_item| {
let (Some(group_name), Some(node)) = (&selected_item.name, &selected_item.now) else {
return false;
};
let Some(group) = proxies.proxies.get(group_name.as_str()) else {
return true;
};
let Some(available_nodes) = group.all.as_deref().filter(|nodes| !nodes.is_empty()) else {
return true;
};
!node_is_available(available_nodes, node)
})
}
fn reconcile_selected_nodes(
selected: &[PrfSelected],
previous: Option<&Proxies>,
proxies: &Proxies,
) -> SelectedNodesPlan {
let mut plan = SelectedNodesPlan {
selected: Vec::with_capacity(selected.len()),
activations: Vec::new(),
repaired_count: 0,
};
let mut seen_groups = HashSet::new();
let mut unique_selected = selected
.iter()
.rev()
.filter(|item| item.name.as_ref().is_some_and(|name| seen_groups.insert(name.clone())))
.collect::<Vec<_>>();
unique_selected.reverse();
plan.repaired_count += selected.len() - unique_selected.len();
for selected_item in unique_selected {
let (Some(group_name), Some(node)) = (&selected_item.name, &selected_item.now) else {
plan.repaired_count += 1;
continue;
};
let Some(group) = proxies.proxies.get(group_name.as_str()) else {
if previous.is_some_and(|snapshot| !snapshot.proxies.contains_key(group_name.as_str())) {
plan.repaired_count += 1;
} else {
plan.selected.push(selected_item.clone());
}
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
};
let Some(available_nodes) = group.all.as_deref().filter(|nodes| !nodes.is_empty()) else {
// Provider-backed groups can be temporarily incomplete immediately after a reload.
plan.selected.push(selected_item.clone());
continue;
};
let is_selectable_group = matches!(
&group.proxy_type,
ProxyType::Selector | ProxyType::URLTest | ProxyType::Fallback | ProxyType::LoadBalance
);
if !is_selectable_group {
let preferred_node = group
.now
.as_deref()
.filter(|current| node_is_available(available_nodes, current))
.or_else(|| node_is_available(available_nodes, node).then_some(node.as_str()));
if let Some(preferred_node) = preferred_node {
if preferred_node != node.as_str() {
plan.repaired_count += 1;
}
plan.selected.push(PrfSelected {
name: Some(group_name.clone()),
now: Some(preferred_node.into()),
});
} else {
plan.repaired_count += 1;
}
continue;
}
if !is_mihomo_ready {
logging!(
error,
if node_is_available(available_nodes, node) {
plan.selected.push(selected_item.clone());
if group.now.as_deref() != Some(node.as_str()) {
plan.activations.push((group_name.clone(), node.clone()));
}
continue;
}
let missing_was_confirmed = previous
.and_then(|snapshot| snapshot.proxies.get(group_name.as_str()))
.and_then(|group| group.all.as_deref())
.filter(|nodes| !nodes.is_empty())
.is_some_and(|nodes| !node_is_available(nodes, node));
if !missing_was_confirmed {
plan.selected.push(selected_item.clone());
continue;
}
plan.repaired_count += 1;
if let Some(current_node) = group
.now
.as_deref()
.filter(|current| node_is_available(available_nodes, current))
{
plan.selected.push(PrfSelected {
name: Some(group_name.clone()),
now: Some(current_node.into()),
});
}
}
plan
}
fn is_activation_current(generation: u64) -> bool {
ACTIVATE_SELECTED_GENERATION.load(Ordering::Acquire) == generation
}
async fn fetch_proxies_with_timeout() -> Result<Proxies> {
tokio::time::timeout(MIHOMO_OPERATION_TIMEOUT, async {
loop {
match handle::Handle::mihomo().await.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;
}
}
}
})
.await
.context("timed out while waiting for mihomo proxies")
}
async fn select_node_with_timeout(group_name: &String, node: &String) -> Result<()> {
tokio::time::timeout(MIHOMO_OPERATION_TIMEOUT, async {
handle::Handle::mihomo()
.await
.select_node_for_group(group_name, node)
.await
})
.await
.with_context(|| format!("timed out while selecting node [{node}] for group [{group_name}]"))?
.with_context(|| format!("failed to select node [{node}] for group [{group_name}]"))
}
async fn persist_reconciled_selected(
profile_uid: &String,
original_selected: &[PrfSelected],
selected: Vec<PrfSelected>,
generation: u64,
) -> Result<()> {
if !is_activation_current(generation) {
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));
}
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(())
}
async fn activate_selected_nodes_worker(
profile_uid: String,
selected: Vec<PrfSelected>,
generation: u64,
) -> Result<()> {
let first_snapshot = fetch_proxies_with_timeout().await?;
if !is_activation_current(generation) {
return Ok(());
}
let (previous_snapshot, current_snapshot) = if selected_nodes_need_confirmation(&selected, &first_snapshot) {
tokio::time::sleep(SELECTED_NODES_RECHECK_DELAY).await;
if !is_activation_current(generation) {
return Ok(());
}
let current_snapshot = fetch_proxies_with_timeout().await?;
(Some(first_snapshot), current_snapshot)
} else {
(None, first_snapshot)
};
if !is_activation_current(generation) {
return Ok(());
}
let plan = reconcile_selected_nodes(&selected, previous_snapshot.as_ref(), &current_snapshot);
logging!(debug, Type::Config, "selected nodes activation plan: {plan:?}");
for (group_name, node) in &plan.activations {
if !is_activation_current(generation) {
return Ok(());
}
match select_node_with_timeout(group_name, node).await {
Ok(()) => logging!(
info,
Type::Config,
"check that the mihomo api reaches the maximum number of retries, maybe mihomo core is not running"
);
"Selected node for proxy: {group_name}, node: {node}"
),
Err(err) => logging!(error, Type::Config, "{err:#}"),
}
}
if plan.repaired_count > 0 && is_activation_current(generation) {
logging!(
info,
Type::Config,
"repairing {} invalid selected node record(s) for profile {profile_uid}",
plan.repaired_count
);
persist_reconciled_selected(&profile_uid, &selected, plan.selected, generation).await?;
}
if is_activation_current(generation) {
handle::Handle::refresh_clash();
}
Ok(())
}
pub fn activate_selected_nodes() -> Result<()> {
logging!(info, Type::Config, "starting activating selected nodes");
let mut active_task = ACTIVATE_SELECTED_TASK.lock();
let generation = ACTIVATE_SELECTED_GENERATION.fetch_add(1, Ordering::AcqRel) + 1;
let previous_task = active_task.take();
let handle = tokio::spawn(async move {
if let Some(previous_task) = previous_task {
let _ = previous_task.await;
}
if !is_activation_current(generation) {
return;
}
if let Some(selected) = profile.selected.as_ref() {
logging!(debug, Type::Config, "selected nodes: {selected:?}");
for selected_item in selected {
let mut retry = 10;
if let Some(group_name) = selected_item.name.as_ref()
&& let Some(node) = selected_item.now.as_ref()
{
while retry >= 0 {
logging!(debug, Type::Config, "check node[{node}] exists");
if mihomo.get_proxy_by_name(node).await.is_ok() {
logging!(debug, Type::Config, "node[{node}] exists");
break;
}
retry -= 1;
tokio::time::sleep(Duration::from_secs(1)).await;
}
if retry < 0 {
logging!(
error,
Type::Config,
"Failed to select node for proxy: {group_name}, node: {node}, because the node [{node}] does not exist"
);
continue;
}
if mihomo.select_node_for_group(group_name, node).await.is_err() {
logging!(
error,
Type::Config,
"Failed to select node for proxy: {group_name}, node: {node}"
);
} else {
logging!(
info,
Type::Config,
"Selected node for proxy: {group_name}, node: {node}"
);
}
}
let result = async {
let profiles = Config::profiles().await.latest_arc();
let current = profiles.get_current().context("no current profile running")?.clone();
let selected = profiles
.get_item(&current)
.context("failed to get current profile")?
.selected
.clone()
.unwrap_or_default();
if selected.is_empty() {
return Ok(());
}
// refresh clash
handle::Handle::refresh_clash();
activate_selected_nodes_worker(current, selected, generation).await
}
.await;
if is_activation_current(generation) {
if let Err(err) = result {
logging!(error, Type::Config, "failed to activate selected nodes: {err:#}");
}
logging!(info, Type::Config, "activating selected nodes done!");
}
logging!(info, Type::Config, "activating selected nodes done!");
*ACTIVATE_SELECTED_TASK.lock() = None;
});
*ACTIVATE_SELECTED_TASK.lock() = Some(handle);
*active_task = Some(handle);
drop(active_task);
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
use tauri_plugin_mihomo::models::Proxy;
fn selected(group: &str, node: &str) -> PrfSelected {
PrfSelected {
name: Some(group.into()),
now: Some(node.into()),
}
}
fn proxies(groups: Vec<(&str, &[&str], Option<&str>)>) -> Proxies {
Proxies {
proxies: groups
.into_iter()
.map(|(name, all, now)| {
(
name.to_owned(),
Proxy {
name: name.to_owned(),
all: Some(all.iter().map(|node| (*node).to_owned()).collect()),
now: now.map(str::to_owned),
proxy_type: ProxyType::Selector,
..Proxy::default()
},
)
})
.collect::<HashMap<_, _>>(),
}
}
#[test]
fn keeps_valid_selection_and_activates_when_needed() {
let saved = vec![selected("group", "saved")];
let plan = reconcile_selected_nodes(
&saved,
None,
&proxies(vec![("group", &["current", "saved"], Some("current"))]),
);
assert_eq!(plan.selected, saved);
assert_eq!(plan.activations, vec![("group".into(), "saved".into())]);
assert_eq!(plan.repaired_count, 0);
}
#[test]
fn replaces_missing_node_with_valid_current_node() {
let snapshot = proxies(vec![("group", &["current"], Some("current"))]);
let plan = reconcile_selected_nodes(&[selected("group", "renamed-node")], Some(&snapshot), &snapshot);
assert_eq!(plan.selected, vec![selected("group", "current")]);
assert!(plan.activations.is_empty());
assert_eq!(plan.repaired_count, 1);
}
#[test]
fn validates_membership_in_group_not_global_existence() {
let snapshot = proxies(vec![
("group", &["current"], Some("current")),
("other-node", &[], None),
]);
let plan = reconcile_selected_nodes(&[selected("group", "other-node")], Some(&snapshot), &snapshot);
assert_eq!(plan.selected, vec![selected("group", "current")]);
assert!(plan.activations.is_empty());
assert_eq!(plan.repaired_count, 1);
}
#[test]
fn does_not_activate_non_selectable_groups() {
let snapshot = Proxies {
proxies: HashMap::from([(
"group".to_owned(),
Proxy {
name: "group".to_owned(),
all: Some(vec!["current".to_owned(), "saved".to_owned()]),
now: Some("current".to_owned()),
proxy_type: ProxyType::Direct,
..Proxy::default()
},
)]),
};
let plan = reconcile_selected_nodes(&[selected("group", "saved")], None, &snapshot);
assert_eq!(plan.selected, vec![selected("group", "current")]);
assert!(plan.activations.is_empty());
assert_eq!(plan.repaired_count, 1);
}
#[test]
fn removes_selection_when_group_or_fallback_is_invalid() {
let snapshot = proxies(vec![("group", &["valid"], Some("invalid-current"))]);
let plan = reconcile_selected_nodes(
&[
selected("missing-group", "node"),
selected("group", "missing-node"),
PrfSelected::default(),
],
Some(&snapshot),
&snapshot,
);
assert!(plan.selected.is_empty());
assert!(plan.activations.is_empty());
assert_eq!(plan.repaired_count, 3);
}
#[test]
fn preserves_selection_until_missing_node_is_confirmed() {
let saved = vec![selected("group", "saved")];
let incomplete = proxies(vec![("group", &[], None)]);
let complete = proxies(vec![("group", &["current"], Some("current"))]);
let incomplete_plan = reconcile_selected_nodes(&saved, None, &incomplete);
let one_snapshot_plan = reconcile_selected_nodes(&saved, None, &complete);
assert_eq!(incomplete_plan.selected, saved);
assert_eq!(incomplete_plan.repaired_count, 0);
assert_eq!(one_snapshot_plan.selected, saved);
assert_eq!(one_snapshot_plan.repaired_count, 0);
}
#[test]
fn recovers_when_group_appears_in_second_snapshot() {
let saved = vec![selected("group", "saved")];
let incomplete = Proxies::default();
let complete = proxies(vec![("group", &["current", "saved"], Some("current"))]);
let plan = reconcile_selected_nodes(&saved, Some(&incomplete), &complete);
assert_eq!(plan.selected, saved);
assert_eq!(plan.activations, vec![("group".into(), "saved".into())]);
assert_eq!(plan.repaired_count, 0);
}
#[test]
fn keeps_last_selection_for_duplicate_group_entries() {
let saved = vec![selected("group", "old"), selected("group", "new")];
let snapshot = proxies(vec![("group", &["old", "new"], Some("old"))]);
let plan = reconcile_selected_nodes(&saved, None, &snapshot);
assert_eq!(plan.selected, vec![selected("group", "new")]);
assert_eq!(plan.activations, vec![("group".into(), "new".into())]);
assert_eq!(plan.repaired_count, 1);
}
}