mirror of
https://github.com/clash-verge-rev/clash-verge-rev.git
synced 2026-10-20 09:02:07 +08:00
fix: serialize concurrent configuration updates
This commit is contained in:
@@ -23,7 +23,7 @@ use std::{
|
||||
path::PathBuf,
|
||||
sync::atomic::{AtomicBool, Ordering},
|
||||
};
|
||||
use tokio::sync::OnceCell;
|
||||
use tokio::sync::{Mutex, MutexGuard, OnceCell};
|
||||
use tokio::time::sleep;
|
||||
|
||||
pub struct Config {
|
||||
@@ -34,6 +34,7 @@ pub struct Config {
|
||||
}
|
||||
|
||||
static TUN_SESSION_SUPPRESSED: AtomicBool = AtomicBool::new(false);
|
||||
static CONFIG_WRITE_LOCK: Mutex<()> = Mutex::const_new(());
|
||||
|
||||
impl Config {
|
||||
pub async fn global() -> &'static Self {
|
||||
@@ -66,6 +67,11 @@ impl Config {
|
||||
Self::global().await.runtime_config.clone()
|
||||
}
|
||||
|
||||
/// Serialize transactions that share configuration draft layers.
|
||||
pub(crate) async fn lock_config_write() -> MutexGuard<'static, ()> {
|
||||
CONFIG_WRITE_LOCK.lock().await
|
||||
}
|
||||
|
||||
/// 初始化订阅
|
||||
pub async fn init_config() -> Result<()> {
|
||||
Self::init_config_before_window().await?;
|
||||
|
||||
@@ -47,6 +47,7 @@ impl Config {
|
||||
}
|
||||
|
||||
async fn resolve_startup_mixed_port_inner() -> Result<bool> {
|
||||
let _config_write = Self::lock_config_write().await;
|
||||
let clash = Self::clash().await.latest_arc();
|
||||
let verge = Self::verge().await.latest_arc();
|
||||
let selected_port = clash.get_mixed_port();
|
||||
@@ -89,7 +90,7 @@ impl Config {
|
||||
let clash = Self::clash().await;
|
||||
let verge = Self::verge().await;
|
||||
let runtime = Self::runtime().await;
|
||||
// Every failure below leaves the three layers as they were, without saying so.
|
||||
// Roll back all three drafts if fallback fails.
|
||||
let transaction = DraftTransaction::begin(vec![&clash, &verge, &runtime])?;
|
||||
|
||||
clash.edit_draft(|draft| {
|
||||
@@ -131,9 +132,7 @@ impl Config {
|
||||
}
|
||||
.await;
|
||||
|
||||
// Files are not part of the transaction, so a failed write is restored explicitly.
|
||||
// The drafts are rolled back first: restoring reads the committed configuration, and
|
||||
// must not see a candidate that is being abandoned.
|
||||
// Restore files after rollback so regeneration sees committed values.
|
||||
if let Err(error) = persist_result {
|
||||
transaction.rollback();
|
||||
return match restore_files(&snapshots).await {
|
||||
@@ -210,7 +209,7 @@ fn report_fallback_error(message: String) {
|
||||
});
|
||||
}
|
||||
|
||||
// Only service-managed cores can survive into this startup phase; this app has not spawned its sidecar yet.
|
||||
// Only a service-managed core can still be running during startup.
|
||||
async fn owned_service_core_uses_port(port: u16) -> bool {
|
||||
if !matches!(SERVICE_MANAGER.current().await, ServiceStatus::Ready) {
|
||||
return false;
|
||||
@@ -261,8 +260,7 @@ async fn owned_service_core_uses_port(port: u16) -> bool {
|
||||
|
||||
fn configured_listener_ports(clash: &IClashTemp, verge: &IVerge) -> HashSet<u16> {
|
||||
let mut ports = HashSet::new();
|
||||
// Every listener except the Mixed Port itself: that is the one being reassigned, so
|
||||
// colliding with its current value is exactly what we are trying to do.
|
||||
// Exclude the Mixed Port because it is being reassigned.
|
||||
for key in proxy_listener_keys().filter(|key| *key != MIXED_PORT_KEY) {
|
||||
if let Some(port) = mapping_port(&clash.0, key) {
|
||||
ports.insert(port);
|
||||
|
||||
@@ -8,6 +8,7 @@ use bitflags::bitflags;
|
||||
use clash_verge_draft::{DraftTransaction, SharedDraft};
|
||||
use clash_verge_logging::{Type, logging, logging_error};
|
||||
use serde_yaml_ng::Mapping;
|
||||
use tokio::sync::MutexGuard;
|
||||
|
||||
/// Patch Clash configuration
|
||||
pub async fn patch_clash(patch: &Mapping) -> Result<()> {
|
||||
@@ -265,19 +266,9 @@ async fn process_terminated_flags(update_flags: UpdateFlags, patch: &IVerge) ->
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Apply a patch to the app's configuration, then re-check anything it can invalidate.
|
||||
/// Apply a patch, then reconcile TUN when its setting changes.
|
||||
///
|
||||
/// Today that is TUN, which is a question about the *setting* rather than about the Run State:
|
||||
/// no Run State transition follows a TUN patch, so the reconciliation that reacts to those
|
||||
/// would never see it. Switching TUN on from the global hotkey is the case that reaches here —
|
||||
/// unlike the tray item and the settings switch it is not gated on availability, and the flags
|
||||
/// the patch raises are answered by a Core reload that changes no Run State at all. Left alone
|
||||
/// the setting stays on and every surface reports TUN as enabled while nothing carries its
|
||||
/// traffic.
|
||||
///
|
||||
/// The reconciliation writes configuration of its own, so it goes through
|
||||
/// [`apply_verge_patch`] rather than back through here. That makes the absence of a cycle a
|
||||
/// property of the call graph rather than of a runtime early-return.
|
||||
/// TUN patches do not always produce a Run State transition, so reconciliation is explicit.
|
||||
pub async fn patch_verge(patch: &IVerge, not_save_file: bool) -> Result<()> {
|
||||
apply_verge_patch(patch, not_save_file).await?;
|
||||
if patch.enable_tun_mode.is_some() {
|
||||
@@ -286,18 +277,21 @@ pub async fn patch_verge(patch: &IVerge, not_save_file: bool) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Apply a patch and nothing else. For callers that are themselves a reconciliation.
|
||||
/// 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;
|
||||
apply_verge_patch_locked(&config_write, patch, not_save_file).await
|
||||
}
|
||||
|
||||
/// Apply a patch with the shared configuration write lock already held.
|
||||
/// Callers must pass the guard returned by [`Config::lock_config_write`].
|
||||
pub(super) async fn apply_verge_patch_locked(
|
||||
_config_write: &MutexGuard<'_, ()>,
|
||||
patch: &IVerge,
|
||||
not_save_file: bool,
|
||||
) -> Result<()> {
|
||||
let verge = Config::verge().await;
|
||||
// Applying the flags can fail, and until now that `?` returned straight out of here past
|
||||
// an `if let Err(..) { discard() }` the compiler could never reach — leaving the failed
|
||||
// edit sitting in the draft, where every later reader saw a value that was never applied
|
||||
// and never written to disk.
|
||||
//
|
||||
// Claiming the layer is what makes the rollback safe. This function holds its draft across
|
||||
// `process_terminated_flags`, which restarts the Core and can take seconds; a second patch
|
||||
// arriving in that window used to share the one draft slot, and whichever of the two failed
|
||||
// first discarded the other's staged edit too.
|
||||
// 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));
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@ pub async fn probe_listener(request: ListenerProbe) -> Result<ListenerProbeOutco
|
||||
#[allow(clippy::cognitive_complexity)]
|
||||
pub async fn save_proxy_ports(settings: ProxyPortSettings) -> Result<SaveProxyPortsOutcome> {
|
||||
settings.validate()?;
|
||||
let _config_write = Config::lock_config_write().await;
|
||||
let manager = CoreManager::global();
|
||||
if !manager.try_start_config_update() {
|
||||
bail!("configuration update is already running");
|
||||
@@ -38,27 +39,19 @@ pub async fn save_proxy_ports(settings: ProxyPortSettings) -> Result<SaveProxyPo
|
||||
manager.finish_config_update();
|
||||
}
|
||||
|
||||
// Read before opening the transaction: rolling back discards whatever draft each layer
|
||||
// holds, so a transaction opened before this `?` would throw away drafts staged by
|
||||
// someone else that this function never touched.
|
||||
// Read first because transaction rollback discards all claimed drafts.
|
||||
let current = current_runtime_mapping().await?;
|
||||
|
||||
let clash = Config::clash().await;
|
||||
let verge = Config::verge().await;
|
||||
let runtime = Config::runtime().await;
|
||||
// Every rejection and every failure below leaves the three layers as they were.
|
||||
// Roll back all three drafts if the candidate is rejected or fails.
|
||||
let transaction = DraftTransaction::begin(vec![&clash, &verge, &runtime])?;
|
||||
|
||||
stage_proxy_ports(&settings).await;
|
||||
// The candidate ports are staged but nothing is serving them yet, so close PAC rather
|
||||
// than hand out a script for a port that is between owners. The PAC endpoint resolves the
|
||||
// Mixed Port through the draft layer, so from here until the drafts are committed or rolled
|
||||
// back it would otherwise answer with a port nothing is listening on.
|
||||
// Hide PAC while the staged ports are not yet serving traffic.
|
||||
manager.core_starting();
|
||||
// Most ways out of here are a rejection: the candidate is refused and the Core keeps
|
||||
// serving on its old ports, having never been stopped. Re-deriving PAC from the Running
|
||||
// Mode is what reopens it for that Core — and it is equally correct after a restart that
|
||||
// succeeded, or one that failed and left the Core down.
|
||||
// Recompute PAC state after every exit path.
|
||||
defer! {
|
||||
manager.core_start_settled();
|
||||
}
|
||||
@@ -82,8 +75,7 @@ pub async fn save_proxy_ports(settings: ProxyPortSettings) -> Result<SaveProxyPo
|
||||
}
|
||||
|
||||
let snapshots = capture_config_files().await?;
|
||||
// From here the drafts alone are no longer enough to undo things: files are on disk and
|
||||
// the core may be restarted, so failures restore explicitly as well as rolling back.
|
||||
// Once files change, failures must restore both files and drafts.
|
||||
if let Err(error) = Config::generate_file(ConfigType::Run).await {
|
||||
transaction.rollback();
|
||||
return match restore_files(&snapshots).await {
|
||||
@@ -210,10 +202,7 @@ async fn persist_proxy_port_sources() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Roll the drafts back *now*, rather than on the way out.
|
||||
///
|
||||
/// The transaction discards when it drops, but rollback has to regenerate files from the
|
||||
/// committed configuration first, so the drafts must be gone before that happens.
|
||||
/// Discard candidate drafts before restoring files from committed configuration.
|
||||
async fn discard_proxy_port_drafts() {
|
||||
Config::clash().await.discard();
|
||||
Config::verge().await.discard();
|
||||
|
||||
+20
-43
@@ -1,12 +1,7 @@
|
||||
//! Turning TUN off when it stops being possible.
|
||||
//!
|
||||
//! The decision is [`RunState::tun_should_be_disabled`]; this is the action it triggers. They
|
||||
//! are separate because the decision is a question about Service capability, which Run State
|
||||
//! owns, while writing user configuration is not something Run State should do.
|
||||
//! Disables TUN when the current Run State cannot support it.
|
||||
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
use clash_verge_draft::DraftBusy;
|
||||
use clash_verge_logging::{Type, logging};
|
||||
use scopeguard::defer;
|
||||
|
||||
@@ -15,48 +10,41 @@ use crate::{
|
||||
core::{handle::Handle, manager::RunningMode, runstate::RUN_STATE},
|
||||
};
|
||||
|
||||
/// Set while a disable is being written.
|
||||
/// Set while an automatic TUN disable is being written.
|
||||
///
|
||||
/// Patching the config produces further Run State transitions, each of which reconciles again;
|
||||
/// without this the first loss of capability would start an unbounded cascade. Cleared on the
|
||||
/// way out however that happens — an early return, an error, or a panic — because a flag stuck
|
||||
/// on would silence every later reconciliation for the life of the process.
|
||||
/// Drops new requests during the write to prevent recursive retries. No-op reconciliations leave
|
||||
/// it clear so later state changes can still be processed.
|
||||
static DISABLING_TUN: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
/// Turn TUN off if the Run State says it cannot work.
|
||||
///
|
||||
/// Reads the current Run State rather than taking a snapshot: this is invoked from a spawned
|
||||
/// task, so whatever transition triggered it may be several states out of date by the time it
|
||||
/// runs, and acting on a stale "cannot work" is how a transient startup state would silently
|
||||
/// rewrite the user's configuration.
|
||||
///
|
||||
/// A no-op in every case except a settled one, so it is safe to call on every transition.
|
||||
/// Disable TUN when the current settled Run State cannot support it.
|
||||
pub async fn reconcile_tun_availability() {
|
||||
// Coalesce all reconciliation requests while a disable write is in flight.
|
||||
if DISABLING_TUN.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Wait for any config transaction, then decide from fresh state and committed config.
|
||||
let config_write = Config::lock_config_write().await;
|
||||
let state = RUN_STATE.state();
|
||||
|
||||
// Before the Core is running the startup decision has not been taken yet: the Service may
|
||||
// be about to be installed, and `prepare_startup` reads `enable_tun_mode` to decide
|
||||
// whether to ask. Turning TUN off here would both undo the user's setting and remove the
|
||||
// very prompt that would have made it work.
|
||||
// Startup may still install the Service and uses this setting to decide whether to prompt.
|
||||
if matches!(state.mode, RunningMode::NotRunning) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Falling back to Sidecar already suppresses TUN for this session. That is deliberately
|
||||
// not a persistent change — `Config::disable_tun_and_persist` is the separate, explicit
|
||||
// act — so honour the decision that was already taken rather than escalating it.
|
||||
// Sidecar fallback suppresses TUN for this session without changing the saved preference.
|
||||
if Config::tun_suppressed_for_session() {
|
||||
return;
|
||||
}
|
||||
|
||||
let tun_enabled = Config::verge().await.latest_arc().enable_tun_mode.unwrap_or(false);
|
||||
// Do not rewrite the saved preference based on an uncommitted draft from a legacy writer.
|
||||
let tun_enabled = Config::verge().await.data_arc().enable_tun_mode.unwrap_or(false);
|
||||
if !state.tun_should_be_disabled(tun_enabled) {
|
||||
return;
|
||||
}
|
||||
|
||||
if DISABLING_TUN.swap(true, Ordering::AcqRel) {
|
||||
return;
|
||||
}
|
||||
// The config lock already serializes writers; the flag only sheds newly spawned work.
|
||||
DISABLING_TUN.store(true, Ordering::Release);
|
||||
defer! {
|
||||
DISABLING_TUN.store(false, Ordering::Release);
|
||||
}
|
||||
@@ -71,20 +59,9 @@ pub async fn reconcile_tun_availability() {
|
||||
..IVerge::default()
|
||||
};
|
||||
|
||||
// The patch-and-reconcile form would call straight back into here; this is the plain apply.
|
||||
match super::apply_verge_patch(&patch, false).await {
|
||||
// Avoid recursively reconciling this internal patch.
|
||||
match super::apply_verge_patch_locked(&config_write, &patch, false).await {
|
||||
Ok(()) => Handle::notice_message("tun_mode::auto_disabled", ""),
|
||||
// Losing the race for the config layer is not a failure the user can act on: this runs
|
||||
// on every Run State transition and again whenever the setting itself is patched, so
|
||||
// the writer that won will be followed by another reconciliation. Telling the user that
|
||||
// turning TUN off "failed" would be both alarming and wrong.
|
||||
Err(error) if error.downcast_ref::<DraftBusy>().is_some() => {
|
||||
logging!(
|
||||
debug,
|
||||
Type::Core,
|
||||
"deferring the TUN reconciliation, configuration is busy: {error:#}"
|
||||
);
|
||||
}
|
||||
Err(error) => {
|
||||
logging!(error, Type::Core, "failed to turn TUN mode off: {error:#}");
|
||||
Handle::notice_message("tun_mode::auto_disable_failed", "");
|
||||
|
||||
Reference in New Issue
Block a user