diff --git a/Changelog.md b/Changelog.md index b837df4e2..d9ef0a071 100644 --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,8 @@ ## v2.5.4 +> [!IMPORTANT] +> 服务模式下，规则集、代理集合等由内核下载的文件会在内核启动和更新后同步回配置目录。同步的是副本，修改副本不影响运行中的内核；需要手工维护的集合请使用 `type: file`。 +
🐞 修复问题 @@ -7,7 +10,7 @@ - 修复代理守卫反复失败后无限重试且不告知用户的问题，连续失败后停止并告知 - 修复其余系统代理问题：未开启过的用户每次启动收到报错、恢复备份后未关闭 - 修复服务健康探测结果粘滞、受限机器上安装服务后无法切换服务模式，失败时回退 Sidecar 且开关状态与内核一致 -- 修复服务模式下升级内核后重启仍运行旧版本的问题（需要 clash-verge-service 2.6.8） +- 修复服务模式下升级内核后重启仍运行旧版本的问题 - 修复 Mihomo API 变更后 proxy-provider 被误过滤，节点显示 Unknown 及延迟测试失败 - 修复代理组节点 `PASS-RULE` 被过滤丢失的问题 - 修复内核启动后未恢复上次选择的节点、代理组回落到首个节点的问题 @@ -23,6 +26,7 @@ - 修复订阅与 DNS 编辑器问题：空名称节点导致崩溃、YAML 错误时丢失原始内容、DNS 编辑初始值覆盖用户配置 - 修复关闭「自动检查更新」后首页手动检查更新直接提示已是最新版本的问题 - 改进错误提示：显示完整原因而非操作系统原文、不再出现 [object Object]、过长时截断（右键复制完整内容） +- 修复服务模式下规则集、代理集合文件在配置目录中消失的问题 **🖥️ Windows** @@ -71,7 +75,7 @@
🚀 优化改进 -- TUN 模式设置也遵循 原始配置 < Merge 覆写 < Script 脚本 < GUI +- 优化 TUN 设置优先级：原始配置 < Merge 覆写 < Script 脚本 < GUI，GUI 仅接管其编辑的字段 - 增强代理页刷新、链路选择与异常状态下的稳定性 - 加强服务多用户隔离与身份认证，新增旧版服务迁移引导 - 系统代理写入半途失败后，系统状态、代理守卫与配置保持一致 diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 611cba083..6b958ec24 100755 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -99,7 +99,7 @@ tracing = { workspace = true } tauri-plugin-devtools = { version = "2.1.0", optional = true } tauri-plugin-mihomo = { git = "https://github.com/clash-verge-rev/tauri-plugin-mihomo", branch = "main" } async-trait = "0.1.92" -clash_verge_service_ipc = { git = "https://github.com/clash-verge-rev/clash-verge-service-ipc.git", tag = "v2.6.8", version = "2.6.8", features = [ +clash_verge_service_ipc = { git = "https://github.com/clash-verge-rev/clash-verge-service-ipc.git", tag = "v2.6.9", version = "2.6.9", features = [ "client", ] } arc-swap = "1.9.2" diff --git a/src-tauri/src/cmd/service.rs b/src-tauri/src/cmd/service.rs index 4d5aed4d4..1d1a21074 100644 --- a/src-tauri/src/cmd/service.rs +++ b/src-tauri/src/cmd/service.rs @@ -1,7 +1,11 @@ use super::{CmdResult, proxy_aware_coded_error}; -use crate::core::{ - CoreManager, - service::{SERVICE_MANAGER, ServiceStatus}, +use crate::{ + constants::timing, + core::{ + CoreManager, + manager::RunningMode, + service::{SERVICE_MANAGER, ServiceStatus, request_runtime_provider_sync}, + }, }; async fn execute_service_operation_sync(status: ServiceStatus, error_code: &str) -> CmdResult { @@ -52,3 +56,10 @@ pub async fn continue_with_sidecar() -> CmdResult { .await .map_err(|error| proxy_aware_coded_error(&error, "SERVICE_SIDECAR_FAILED")) } + +#[tauri::command] +pub fn sync_runtime_providers() { + if matches!(*CoreManager::global().get_running_mode(), RunningMode::Service) { + request_runtime_provider_sync(timing::RUNTIME_PROVIDER_SETTLE); + } +} diff --git a/src-tauri/src/constants.rs b/src-tauri/src/constants.rs index 251a14f96..f629c405f 100644 --- a/src-tauri/src/constants.rs +++ b/src-tauri/src/constants.rs @@ -24,6 +24,11 @@ pub mod timing { // milliseconds; anything slower means the Service is still working, so ReplaceCore beats waiting. pub const STAGE_CONFIRM_TIMEOUT: Duration = Duration::from_secs(5); + pub const RUNTIME_PROVIDER_SYNC_DELAY: Duration = Duration::from_secs(15); + pub const RUNTIME_PROVIDER_SYNC_RETRY_DELAY: Duration = Duration::from_secs(15); + // mihomo rewrites provider caches in place. + pub const RUNTIME_PROVIDER_SETTLE: Duration = Duration::from_secs(2); + // Windows 服务冷启动较慢,避免过早回退 sidecar。 #[cfg(target_os = "windows")] pub const SERVICE_WAIT_MAX: Duration = Duration::from_millis(30000); @@ -59,4 +64,18 @@ pub mod tun { pub const DEFAULT_STACK: &str = "gvisor"; pub const DNS_HIJACK: &[&str] = &["any:53"]; + + /// Saved dialog fields; `enable` belongs to the verge switch. + pub const GUI_KEYS: &[&str] = &[ + "stack", + "device", + "auto-route", + "route-exclude-address", + #[cfg(target_os = "linux")] + "auto-redirect", + "auto-detect-interface", + "dns-hijack", + "strict-route", + "mtu", + ]; } diff --git a/src-tauri/src/core/manager/config.rs b/src-tauri/src/core/manager/config.rs index d6324e553..2f3e7f7f0 100644 --- a/src-tauri/src/core/manager/config.rs +++ b/src-tauri/src/core/manager/config.rs @@ -345,6 +345,7 @@ impl CoreManager { ConfigApplication::ReloadFrom(staged) => match self.reload_config(&staged).await { Ok(()) => { tracing::Span::current().record("outcome", "staged"); + crate::core::service::request_runtime_provider_sync(timing::RUNTIME_PROVIDER_SYNC_DELAY); return Ok(()); } Err(err) => logging!( diff --git a/src-tauri/src/core/manager/state.rs b/src-tauri/src/core/manager/state.rs index be2d7a73a..d7486651f 100644 --- a/src-tauri/src/core/manager/state.rs +++ b/src-tauri/src/core/manager/state.rs @@ -276,6 +276,7 @@ impl CoreManager { self.mark_core_ready(); self.core_started(RunningMode::Service); self.restore_selected_nodes().await; + service::request_runtime_provider_sync(crate::constants::timing::RUNTIME_PROVIDER_SYNC_DELAY); return Ok(()); } Err(e) => { @@ -300,6 +301,7 @@ impl CoreManager { self.mark_core_ready(); self.core_started(RunningMode::Service); self.restore_selected_nodes().await; + service::request_runtime_provider_sync(crate::constants::timing::RUNTIME_PROVIDER_SYNC_DELAY); Ok(()) } } diff --git a/src-tauri/src/core/runtime_bundle.rs b/src-tauri/src/core/runtime_bundle.rs index ede83e520..3569100e0 100644 --- a/src-tauri/src/core/runtime_bundle.rs +++ b/src-tauri/src/core/runtime_bundle.rs @@ -1,7 +1,7 @@ use anyhow::{Context as _, Result, bail}; use clash_verge_logging::{Type, logging}; use clash_verge_service_ipc::{RemoteProvider, RuntimeAsset, RuntimeBundle}; -use serde_yaml_ng::Value; +use serde_yaml_ng::{Mapping, Value}; use std::collections::HashSet; use std::path::{Component, Path, PathBuf}; @@ -13,15 +13,47 @@ pub(crate) const GEO_ASSETS: &[&str] = &[ "GeoSite.dat", ]; +/// Provider identity used by the runtime config and core API. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RemoteProviderRef { + pub section: &'static str, + pub name: String, + pub provider: RemoteProvider, +} + +struct RuntimeParts { + config: Value, + assets: Vec, + remote_providers: Vec, +} + pub(crate) async fn collect_runtime_bundle(config_file: &Path, core_path: &Path) -> Result { let yaml = tokio::fs::read_to_string(config_file) .await .with_context(|| format!("failed to read runtime config {config_file:?}"))?; - let mut config: Value = + let config: Value = serde_yaml_ng::from_str(&yaml).with_context(|| format!("failed to parse runtime config {config_file:?}"))?; let config_root = config_file .parent() .ok_or_else(|| anyhow::anyhow!("runtime config has no parent directory"))?; + let parts = collect_runtime_parts(config, config_root)?; + Ok(RuntimeBundle { + yaml: serde_yaml_ng::to_string(&parts.config).context("failed to serialize service runtime config")?, + assets: parts.assets, + remote_providers: parts + .remote_providers + .into_iter() + .map(|declared| declared.provider) + .collect(), + core_path: core_path.to_string_lossy().into_owned(), + }) +} + +pub(crate) fn remote_providers_of(config: &Mapping, config_root: &Path) -> Result> { + Ok(collect_runtime_parts(Value::Mapping(config.clone()), config_root)?.remote_providers) +} + +fn collect_runtime_parts(mut config: Value, config_root: &Path) -> Result { let config_root = std::fs::canonicalize(config_root)?; let mut assets = Vec::new(); let mut destinations = HashSet::new(); @@ -53,11 +85,10 @@ pub(crate) async fn collect_runtime_bundle(config_file: &Path, core_path: &Path) } } - Ok(RuntimeBundle { - yaml: serde_yaml_ng::to_string(&config).context("failed to serialize service runtime config")?, + Ok(RuntimeParts { + config, assets, remote_providers, - core_path: core_path.to_string_lossy().into_owned(), }) } @@ -65,11 +96,11 @@ pub(crate) async fn collect_runtime_bundle(config_file: &Path, core_path: &Path) /// Remote URLs let the service decide whether a core download cache remains reusable. fn collect_provider_assets( config: &mut Value, - section: &str, + section: &'static str, config_root: &Path, destinations: &mut HashSet, assets: &mut Vec, - remote_providers: &mut Vec, + remote_providers: &mut Vec, ) -> Result<()> { let Some(providers) = config .as_mapping_mut() @@ -95,9 +126,9 @@ fn collect_provider_assets( // Reserve remote destinations too; copied and downloaded files must not collide. match remote_providers .iter() - .find(|declared| declared.destination == destination) + .find(|declared| declared.provider.destination == destination) { - Some(declared) if declared.url == url => {} + Some(declared) if declared.provider.url == url => {} Some(_) => { bail!("runtime provider destination {destination:?} is declared for two different sources") } @@ -105,9 +136,13 @@ fn collect_provider_assets( if !destinations.insert(destination.clone()) { bail!("runtime provider destination {destination:?} is claimed more than once"); } - remote_providers.push(RemoteProvider { - destination: destination.clone(), - url, + remote_providers.push(RemoteProviderRef { + section, + name: name.as_str().unwrap_or_default().to_owned(), + provider: RemoteProvider { + destination: destination.clone(), + url, + }, }); } } diff --git a/src-tauri/src/core/service.rs b/src-tauri/src/core/service.rs index 3c55bfa83..9b3b79662 100644 --- a/src-tauri/src/core/service.rs +++ b/src-tauri/src/core/service.rs @@ -1,7 +1,7 @@ -#[cfg(any(target_os = "macos", target_os = "windows"))] use crate::utils::dirs; use crate::{ - config::Config, + config::{Config, runtime::IRuntime}, + constants, core::{ CoreManager, handle::Handle, @@ -12,27 +12,29 @@ use crate::{ OwnerRecoveryReason, OwnerSample, OwnerStep, OwnerWatch, PendingAction, RUN_STATE, ReadyWaitError, RunState, RunStateEnv, RunStateStore, ServiceHealth, }, - runtime_bundle::collect_runtime_bundle, + runtime_bundle::{RemoteProviderRef, collect_runtime_bundle, remote_providers_of}, tray::Tray, }, process::AsyncHandler, }; -use anyhow::{Context as _, Result, bail}; +use anyhow::{Context as _, Result, anyhow, bail}; +use clash_verge_draft::Draft; use clash_verge_logging::{Type, logging}; use clash_verge_service_ipc::{ - MacosProxyConfig, OwnerSessionProof, ProxyApplyOutcome, RuntimeBundle, ServiceErrorCode, StageRuntimeOutcome, - StartClashRequest, WriterConfig, + MacosProxyConfig, OwnerCredentials, OwnerSessionProof, ProtocolInfo, ProxyApplyOutcome, RuntimeBundle, + RuntimeFileOutcome, RuntimeFileRequest, ServiceErrorCode, StageRuntimeOutcome, StartClashRequest, WriterConfig, }; use once_cell::sync::Lazy; use parking_lot::Mutex; use std::{ borrow::Cow, + collections::HashMap, env::current_exe, future::Future, path::{Path, PathBuf}, process::Command as StdCommand, - sync::atomic::{AtomicU64, Ordering}, - time::Duration, + sync::atomic::{AtomicBool, AtomicU64, Ordering}, + time::{Duration, SystemTime, UNIX_EPOCH}, }; static OWNER_MONITOR_GENERATION: AtomicU64 = AtomicU64::new(0); @@ -44,6 +46,7 @@ static ACTIVE_SERVICE_SESSION: Lazy>> = Lazy: struct ActiveServiceSession { proof: OwnerSessionProof, supports_runtime_staging: bool, + supports_runtime_file_read: bool, } fn generate_service_session_token() -> Result { @@ -68,21 +71,43 @@ pub(crate) fn active_service_supports_runtime_staging() -> bool { .is_some_and(|session| session.supports_runtime_staging) } +pub(crate) fn active_service_supports_runtime_file_read() -> bool { + ACTIVE_SERVICE_SESSION + .lock() + .as_ref() + .is_some_and(|session| session.supports_runtime_file_read) +} + pub(crate) fn clear_active_service_session() { ACTIVE_SERVICE_SESSION.lock().take(); } -/// Probes staging support without failing startup when the fast path is unavailable. +#[derive(Clone, Copy, Default)] +struct ServiceCapabilities { + runtime_staging: bool, + runtime_file_read: bool, +} + +impl ServiceCapabilities { + const fn of(info: &ProtocolInfo) -> Self { + Self { + runtime_staging: info.supports_runtime_staging(), + runtime_file_read: info.supports_runtime_file_read(), + } + } +} + +/// Failed capability probes must not block startup. #[tracing::instrument(skip_all, level = "info", fields(supported = tracing::field::Empty))] -async fn probe_runtime_staging_support() -> bool { +async fn probe_service_capabilities() -> ServiceCapabilities { match clash_verge_service_ipc::get_version().await { Ok(response) if response.code == 0 => { - let supported = response + let capabilities = response .data .as_ref() - .is_some_and(clash_verge_service_ipc::ProtocolInfo::supports_runtime_staging); - tracing::Span::current().record("supported", supported); - supported + .map_or_else(ServiceCapabilities::default, ServiceCapabilities::of); + tracing::Span::current().record("supported", capabilities.runtime_staging); + capabilities } Ok(response) => { tracing::Span::current().record("supported", false); @@ -93,7 +118,7 @@ async fn probe_runtime_staging_support() -> bool { response.code, response.message ); - false + ServiceCapabilities::default() } Err(error) => { tracing::Span::current().record("supported", false); @@ -102,7 +127,7 @@ async fn probe_runtime_staging_support() -> bool { Type::Service, "无法查询服务协议版本: {error:#}；配置变更将走重启路径" ); - false + ServiceCapabilities::default() } } } @@ -1060,14 +1085,15 @@ pub(super) async fn start_with_existing_service(config_file: &Path) -> Result<() let result = response.data.context("Clash Verge Service 未返回会话信息")?; tracing::Span::current().record("generation", result.session.generation); - let supports_runtime_staging = probe_runtime_staging_support().await; - tracing::Span::current().record("staging", supports_runtime_staging); + let capabilities = probe_service_capabilities().await; + tracing::Span::current().record("staging", capabilities.runtime_staging); *ACTIVE_SERVICE_SESSION.lock() = Some(ActiveServiceSession { proof: OwnerSessionProof { generation: result.session.generation, token: proposed_session_token, }, - supports_runtime_staging, + supports_runtime_staging: capabilities.runtime_staging, + supports_runtime_file_read: capabilities.runtime_file_read, }); // PAC follows the Running Mode; the caller opens it via `core_started(Service)`. @@ -1138,16 +1164,424 @@ pub(crate) async fn get_clash_log_snapshot_by_service() -> Result { bail!(response.message); } let encoded = response.data.context("服务未返回核心日志快照")?; - if encoded.len() % 2 != 0 { - bail!("服务返回了无效的核心日志快照"); - } - let mut content = Vec::with_capacity(encoded.len() / 2); - for offset in (0..encoded.len()).step_by(2) { - content.push(u8::from_str_radix(&encoded[offset..offset + 2], 16).context("服务返回了无效的核心日志快照")?); - } + let content = decode_hex(&encoded).context("服务返回了无效的核心日志快照")?; Ok(String::from_utf8_lossy(&content).into_owned()) } +fn decode_hex(encoded: &str) -> Result> { + if !encoded.len().is_multiple_of(2) { + bail!("hex payload has an odd length"); + } + (0..encoded.len()) + .step_by(2) + .map(|offset| u8::from_str_radix(&encoded[offset..offset + 2], 16).context("hex payload is malformed")) + .collect() +} + +static PROVIDER_SYNC_QUEUED: AtomicBool = AtomicBool::new(false); +static PROVIDER_SYNC_SERIAL: Lazy> = Lazy::new(|| tokio::sync::Mutex::new(())); +static SYNC_TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0); +const RUNTIME_PROVIDER_SYNC_ATTEMPTS: u32 = 4; +const CONTENT_COMPARE_CHUNK: usize = 64 * 1024; + +pub(crate) fn request_runtime_provider_sync(delay: Duration) { + if PROVIDER_SYNC_QUEUED.swap(true, Ordering::AcqRel) { + return; + } + 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); + } + 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; + } + }); +} + +#[derive(Debug, Default)] +struct ProviderSync { + synced: usize, + pending: usize, +} + +struct FetchedCache { + declared: RemoteProviderRef, + temp: PathBuf, + len: u64, + mtime_ns: Option, +} + +/// Core load timestamps keyed by provider section and name. +type ProviderLoadTimes = HashMap<(&'static str, String), Option>; + +#[tracing::instrument(level = "info", fields(synced = tracing::field::Empty, pending = tracing::field::Empty))] +async fn sync_runtime_providers_by_service() -> Result { + if !active_service_supports_runtime_file_read() { + return Ok(ProviderSync::default()); + } + let credentials = current_owner_credentials()?; + let session = active_service_session()?; + let app_dir = dirs::app_home_dir()?; + let runtime = Config::runtime().await; + let mut outcome = ProviderSync::default(); + + let loaded_before = provider_load_times().await; + let mut fetched = Vec::new(); + for declared in applied_remote_providers(&runtime, &app_dir)? { + let (temp, file) = match create_sync_temp(&app_dir.join(&declared.provider.destination)).await { + Ok(created) => created, + Err(error) => { + logging!(warn, Type::Service, "{error:#}"); + outcome.pending += 1; + continue; + } + }; + match fetch_runtime_file(&credentials, &session, &declared.provider.destination, file).await { + Ok(Fetch::Complete { len, mtime_ns }) => fetched.push(FetchedCache { + declared, + temp, + len, + mtime_ns, + }), + Ok(Fetch::NotReady) => { + outcome.pending += 1; + remove_temp(&temp).await; + } + Ok(Fetch::SessionLost) => { + remove_temp(&temp).await; + discard_fetched(fetched).await; + bail!("the service session ended while provider caches were being read"); + } + Err(error) => { + logging!( + warn, + Type::Service, + "provider cache {} was not read: {error:#}", + declared.provider.destination + ); + outcome.pending += 1; + remove_temp(&temp).await; + } + } + } + + let (loaded_before, loaded_after) = match (loaded_before, provider_load_times().await) { + (Ok(before), Ok(after)) => (before, after), + (Err(error), _) | (_, Err(error)) => { + logging!(warn, Type::Service, "provider caches were not synced: {error:#}"); + outcome.pending += fetched.len(); + discard_fetched(fetched).await; + return Ok(outcome); + } + }; + let mut ready = Vec::new(); + for cache in fetched { + let key = (cache.declared.section, cache.declared.name.clone()); + let updated_at = loaded_after.get(&key).copied().flatten(); + let complete = loaded_before.get(&key) == loaded_after.get(&key) + && finished_loading(cache.mtime_ns, updated_at) + && match recheck_identity(&credentials, &session, &cache).await { + Ok(Fetch::Complete { .. }) => true, + Ok(Fetch::NotReady) | Err(_) => false, + Ok(Fetch::SessionLost) => { + remove_temp(&cache.temp).await; + discard_fetched(ready).await; + bail!("the service session ended while provider caches were being verified"); + } + }; + if complete { + ready.push(cache); + } else { + outcome.pending += 1; + remove_temp(&cache.temp).await; + } + } + + outcome.synced = publish_fetched(&app_dir, &runtime, &session, ready).await?; + tracing::Span::current().record("synced", outcome.synced); + tracing::Span::current().record("pending", outcome.pending); + Ok(outcome) +} + +// The on-disk YAML may describe a rejected apply; use the committed runtime. +fn applied_remote_providers(runtime: &Draft, app_dir: &Path) -> Result> { + let applied = runtime.data_arc(); + match applied.config.as_ref() { + Some(config) => remote_providers_of(config, app_dir), + None => Ok(Vec::new()), + } +} + +// Hold the lifecycle lock through publication to exclude mode and config switches. +async fn publish_fetched( + app_dir: &Path, + runtime: &Draft, + session: &OwnerSessionProof, + fetched: Vec, +) -> Result { + let manager = CoreManager::global(); + let _lifecycle = manager.lifecycle_lock.lock().await; + let unchanged = matches!(*manager.get_running_mode(), RunningMode::Service) + && active_service_session().is_ok_and(|current| current == *session); + let declared = if unchanged { + match applied_remote_providers(runtime, app_dir) { + Ok(declared) => declared, + Err(error) => { + discard_fetched(fetched).await; + return Err(error); + } + } + } else { + Vec::new() + }; + let mut synced = 0; + for cache in fetched { + let target = app_dir.join(&cache.declared.provider.destination); + if !declared.contains(&cache.declared) || same_contents(&cache.temp, &target).await { + remove_temp(&cache.temp).await; + continue; + } + match tokio::fs::rename(&cache.temp, &target).await { + Ok(()) => synced += 1, + Err(error) => { + logging!( + warn, + Type::Service, + "provider cache {} was not published: {error}", + cache.declared.provider.destination + ); + remove_temp(&cache.temp).await; + } + } + } + Ok(synced) +} + +enum Fetch { + Complete { len: u64, mtime_ns: Option }, + NotReady, + SessionLost, +} + +// mihomo rewrites caches in place; reject changing or recently modified files. +async fn fetch_runtime_file( + credentials: &OwnerCredentials, + session: &OwnerSessionProof, + destination: &str, + mut file: tokio::fs::File, +) -> Result { + use tokio::io::AsyncWriteExt as _; + + let mut written = 0_u64; + let mut identity = None; + loop { + let (hex, len, mtime_ns) = match read_chunk(credentials, session, destination, written).await? { + Some(chunk) => chunk, + None => return Ok(Fetch::SessionLost), + }; + let Some((hex, len, mtime_ns)) = hex.map(|hex| (hex, len, mtime_ns)) else { + return Ok(Fetch::NotReady); + }; + if len == 0 || !has_settled(mtime_ns) || *identity.get_or_insert((len, mtime_ns)) != (len, mtime_ns) { + return Ok(Fetch::NotReady); + } + if written == len { + file.flush().await?; + return Ok(Fetch::Complete { len, mtime_ns }); + } + let chunk = decode_hex(&hex).context("服务返回了无效的运行时文件")?; + if chunk.is_empty() { + bail!("the service returned an empty chunk before the end of the file"); + } + file.write_all(&chunk).await?; + written += chunk.len() as u64; + if written > len { + return Ok(Fetch::NotReady); + } + } +} + +async fn recheck_identity( + credentials: &OwnerCredentials, + session: &OwnerSessionProof, + cache: &FetchedCache, +) -> Result { + let destination = &cache.declared.provider.destination; + Ok(match read_chunk(credentials, session, destination, cache.len).await? { + None => Fetch::SessionLost, + Some((Some(hex), len, mtime_ns)) if hex.is_empty() && (len, mtime_ns) == (cache.len, cache.mtime_ns) => { + Fetch::Complete { len, mtime_ns } + } + Some(_) => Fetch::NotReady, + }) +} + +/// Outer `None` means session loss; missing hex means an absent file. +async fn read_chunk( + credentials: &OwnerCredentials, + session: &OwnerSessionProof, + destination: &str, + offset: u64, +) -> Result, u64, Option)>> { + let request = RuntimeFileRequest { + destination: destination.to_owned(), + offset, + }; + let response = clash_verge_service_ipc::read_runtime_file(credentials, session, &request) + .await + .context("无法连接到Clash Verge Service")?; + if response.code == ServiceErrorCode::NotActive as u16 + || response.code == ServiceErrorCode::StaleOwnerSession as u16 + { + return Ok(None); + } + if response.code > 0 { + bail!(response.message); + } + Ok(Some(match response.data.context("服务未返回运行时文件")? { + RuntimeFileOutcome::Absent => (None, 0, None), + RuntimeFileOutcome::Chunk { hex, len, mtime_ns } => (Some(hex), len, mtime_ns), + })) +} + +// updatedAt changes after cache writes; compare it before and after readback. +async fn provider_load_times() -> Result { + let mihomo = Handle::mihomo(); + let (rules, proxies) = tokio::join!(mihomo.get_rule_providers(), mihomo.get_proxy_providers()); + let rules = rules.map_err(|error| anyhow!("failed to query rule providers: {error}"))?; + let proxies = proxies.map_err(|error| anyhow!("failed to query proxy providers: {error}"))?; + let mut loaded = ProviderLoadTimes::new(); + for (name, provider) in rules.providers { + loaded.insert(("rule-providers", name), epoch_nanos(&provider.updated_at)); + } + for (name, provider) in proxies.providers { + loaded.insert( + ("proxy-providers", name), + provider.updated_at.as_deref().and_then(epoch_nanos), + ); + } + Ok(loaded) +} + +fn epoch_nanos(rfc3339: &str) -> Option { + let parsed = chrono::DateTime::parse_from_rfc3339(rfc3339).ok()?; + u64::try_from(parsed.timestamp_nanos_opt()?).ok() +} + +// Allow timestamp granularity; missing load times fall back to settling. +// Windows may retain the old mtime until a writer closes its handle. +fn finished_loading(mtime_ns: Option, updated_at_ns: Option) -> bool { + match (mtime_ns, updated_at_ns) { + (Some(mtime), Some(updated_at)) => { + u128::from(mtime) <= u128::from(updated_at) + constants::timing::RUNTIME_PROVIDER_SETTLE.as_nanos() + } + _ => true, + } +} + +fn has_settled(mtime_ns: Option) -> bool { + let Some(mtime_ns) = mtime_ns else { + return true; + }; + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |since_epoch| since_epoch.as_nanos()); + now.saturating_sub(u128::from(mtime_ns)) >= constants::timing::RUNTIME_PROVIDER_SETTLE.as_nanos() +} + +// Exclusive creation protects existing files from truncation and cleanup. +async fn create_sync_temp(target: &Path) -> Result<(PathBuf, tokio::fs::File)> { + if let Some(parent) = target.parent() { + tokio::fs::create_dir_all(parent).await?; + } + let name = target + .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()) +} + +async fn same_contents(temp: &Path, target: &Path) -> bool { + use tokio::io::AsyncReadExt as _; + + let (Ok(mut left), Ok(mut right)) = (tokio::fs::File::open(temp).await, tokio::fs::File::open(target).await) else { + return false; + }; + let (Ok(left_meta), Ok(right_meta)) = (left.metadata().await, right.metadata().await) else { + return false; + }; + if left_meta.len() != right_meta.len() { + return false; + } + let mut remaining = left_meta.len(); + let mut left_buf = vec![0_u8; CONTENT_COMPARE_CHUNK]; + let mut right_buf = vec![0_u8; CONTENT_COMPARE_CHUNK]; + while remaining > 0 { + let take = usize::try_from(remaining).map_or(CONTENT_COMPARE_CHUNK, |rest| rest.min(CONTENT_COMPARE_CHUNK)); + if left.read_exact(&mut left_buf[..take]).await.is_err() + || right.read_exact(&mut right_buf[..take]).await.is_err() + { + return false; + } + if left_buf[..take] != right_buf[..take] { + return false; + } + remaining -= take as u64; + } + true +} + +async fn remove_temp(temp: &Path) { + let _ = tokio::fs::remove_file(temp).await; +} + +async fn discard_fetched(fetched: Vec) { + for cache in fetched { + remove_temp(&cache.temp).await; + } +} + /// 通过服务停止core #[tracing::instrument(skip_all, level = "info", fields(code = tracing::field::Empty, outcome = tracing::field::Empty))] pub(super) async fn stop_core_by_service() -> Result<()> { diff --git a/src-tauri/src/enhance/mod.rs b/src-tauri/src/enhance/mod.rs index 888597bf2..02dab5842 100644 --- a/src-tauri/src/enhance/mod.rs +++ b/src-tauri/src/enhance/mod.rs @@ -301,7 +301,6 @@ const CONTROL_PLANE_KEYS: &[&str] = &[ "redir-port", #[cfg(target_os = "linux")] "tproxy-port", - "tun", "mode", "allow-lan", "log-level", @@ -317,20 +316,23 @@ const CONTROL_PLANE_KEYS: &[&str] = &[ /// override. As four separate calls that contract lived only in comments. struct AuthoritativeFields { control_plane: Mapping, + tun: Mapping, /// Only tracked when the DNS page owns it; otherwise overrides may set `dns.ipv6` freely. dns_ipv6: Option, } impl AuthoritativeFields { - fn capture(config: &Mapping, enable_dns_settings: bool) -> Self { + fn capture(config: &Mapping, gui_tun_keys: &[Value], enable_dns_settings: bool) -> Self { Self { control_plane: snapshot_control_plane(config), + tun: snapshot_tun(config, gui_tun_keys), dns_ipv6: enable_dns_settings.then(|| snapshot_dns_ipv6(config)).flatten(), } } fn enforce(self, config: Mapping) -> Mapping { let config = enforce_control_plane(config, self.control_plane); + let config = enforce_tun(config, self.tun); enforce_dns_ipv6(config, self.dns_ipv6) } } @@ -359,6 +361,46 @@ fn enforce_control_plane(mut config: Mapping, snapshot: Mapping) -> Mapping { config } +/// Only saved GUI fields and the verge switch override profile settings. +fn gui_tun_keys(clash_config: &Mapping) -> Vec { + let mut keys = vec![Value::from("enable")]; + if let Some(Value::Mapping(tun)) = clash_config.get("tun") { + keys.extend( + constants::tun::GUI_KEYS + .iter() + .filter(|key| tun.contains_key(**key)) + .map(|key| Value::from(*key)), + ); + } + keys +} + +fn snapshot_tun(config: &Mapping, gui_tun_keys: &[Value]) -> Mapping { + let mut snapshot = Mapping::new(); + if let Some(Value::Mapping(tun)) = config.get("tun") { + for key in gui_tun_keys { + if let Some(value) = tun.get(key) { + snapshot.insert(key.clone(), value.clone()); + } + } + } + snapshot +} + +fn enforce_tun(mut config: Mapping, snapshot: Mapping) -> Mapping { + if snapshot.is_empty() { + return config; + } + let mut tun = config + .get("tun") + .and_then(Value::as_mapping) + .cloned() + .unwrap_or_default(); + tun.extend(snapshot); + config.insert(Value::from("tun"), Value::Mapping(tun)); + config +} + /// DNS 页权威的嵌套开关;只在 `enable_dns_settings` 时快照。 fn snapshot_dns_ipv6(config: &Mapping) -> Option { config.get("dns")?.get("ipv6").cloned() @@ -742,6 +784,7 @@ pub async fn enhance( let config = process_seq_items(config, rules_item, proxies_item, groups_item); let exists_keys = use_keys(&config).collect::>(); + let gui_tun_keys = gui_tun_keys(&clash_config); let config = merge_default_config( config, @@ -759,7 +802,7 @@ pub async fn enhance( let config = use_tun(config, enable_tun); let config = apply_dns_settings(config, enable_dns_settings).await; - let authoritative = AuthoritativeFields::capture(&config, enable_dns_settings); + let authoritative = AuthoritativeFields::capture(&config, &gui_tun_keys, enable_dns_settings); let (config, exists_keys, result_map) = process_global_items( config, @@ -1180,7 +1223,7 @@ mod authoritative_field_tests { #[test] fn an_override_cannot_change_a_field_the_app_owns() { let derived = config_with(&[("mode", Value::from("rule")), ("secret", Value::from("ours"))]); - let authoritative = AuthoritativeFields::capture(&derived, false); + let authoritative = AuthoritativeFields::capture(&derived, &[], false); let overridden = config_with(&[("mode", Value::from("global")), ("secret", Value::from("theirs"))]); let result = authoritative.enforce(overridden); @@ -1193,7 +1236,7 @@ mod authoritative_field_tests { fn an_override_cannot_introduce_a_field_the_app_left_out() { // The app decided not to expose the external controller; a profile must not re-add it. let derived = Mapping::new(); - let authoritative = AuthoritativeFields::capture(&derived, false); + let authoritative = AuthoritativeFields::capture(&derived, &[], false); let overridden = config_with(&[("external-controller", Value::from("0.0.0.0:9090"))]); let result = authoritative.enforce(overridden); @@ -1204,7 +1247,7 @@ mod authoritative_field_tests { #[test] fn fields_the_app_does_not_own_survive_an_override() { let derived = config_with(&[("mode", Value::from("rule"))]); - let authoritative = AuthoritativeFields::capture(&derived, false); + let authoritative = AuthoritativeFields::capture(&derived, &[], false); let overridden = config_with(&[("mode", Value::from("global")), ("profile-key", Value::from(1))]); let result = authoritative.enforce(overridden); @@ -1216,7 +1259,7 @@ mod authoritative_field_tests { fn dns_ipv6_is_only_reclaimed_when_the_dns_page_owns_it() { let derived = config_with(&[("dns", dns_with_ipv6(true))]); - let owned = AuthoritativeFields::capture(&derived, true); + let owned = AuthoritativeFields::capture(&derived, &[], true); let restored = owned.enforce(config_with(&[("dns", dns_with_ipv6(false))])); assert_eq!( restored.get(Value::from("dns")).and_then(|dns| dns.get("ipv6")), @@ -1224,7 +1267,7 @@ mod authoritative_field_tests { "with the DNS page on, the app's value wins" ); - let unowned = AuthoritativeFields::capture(&derived, false); + let unowned = AuthoritativeFields::capture(&derived, &[], false); let left_alone = unowned.enforce(config_with(&[("dns", dns_with_ipv6(false))])); assert_eq!( left_alone.get(Value::from("dns")).and_then(|dns| dns.get("ipv6")), @@ -1236,7 +1279,7 @@ mod authoritative_field_tests { #[test] fn restoring_dns_ipv6_never_invents_a_dns_block() { let derived = config_with(&[("dns", dns_with_ipv6(true))]); - let authoritative = AuthoritativeFields::capture(&derived, true); + let authoritative = AuthoritativeFields::capture(&derived, &[], true); // An override removed DNS entirely; reinstating just `ipv6` would be a half-config. let result = authoritative.enforce(Mapping::new()); diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 9d8086a17..19891eba3 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -142,6 +142,7 @@ mod app_init { cmd::reinstall_service, cmd::repair_service, cmd::continue_with_sidecar, + cmd::sync_runtime_providers, cmd::get_clash_info, cmd::patch_clash_config, cmd::patch_clash_mode, diff --git a/src/components/proxy/provider-button.tsx b/src/components/proxy/provider-button.tsx index 84fa690e7..f622f8a43 100644 --- a/src/components/proxy/provider-button.tsx +++ b/src/components/proxy/provider-button.tsx @@ -23,6 +23,7 @@ import { useTranslation } from 'react-i18next' import { updateProxyProvider } from 'tauri-plugin-mihomo-api' import { useAppRefreshers, useProxiesData } from '@/providers/app-data-context' +import { syncRuntimeProviders } from '@/services/cmds' import { showNotice } from '@/services/notice-service' import parseTraffic from '@/utils/parse-traffic' @@ -59,6 +60,7 @@ export const ProviderButton = () => { await updateProxyProvider(name) await refreshProxy() + void syncRuntimeProviders() showNotice.success( 'proxies.feedback.notifications.provider.updateSuccess', @@ -103,6 +105,7 @@ export const ProviderButton = () => { } await refreshProxy() + void syncRuntimeProviders() showNotice.success('proxies.feedback.notifications.provider.allUpdated') } catch (err) { diff --git a/src/components/rule/provider-button.tsx b/src/components/rule/provider-button.tsx index f59f46f28..ad11cc464 100644 --- a/src/components/rule/provider-button.tsx +++ b/src/components/rule/provider-button.tsx @@ -22,6 +22,7 @@ import { useTranslation } from 'react-i18next' import { updateRuleProvider } from 'tauri-plugin-mihomo-api' import { useAppRefreshers, useRulesData } from '@/providers/app-data-context' +import { syncRuntimeProviders } from '@/services/cmds' import { showNotice } from '@/services/notice-service' const TypeBox = styled(Box)<{ component?: React.ElementType }>(({ theme }) => ({ @@ -53,6 +54,7 @@ export const ProviderButton = () => { await refreshRules() await refreshRuleProviders() + void syncRuntimeProviders() showNotice.success( 'rules.feedback.notifications.provider.updateSuccess', @@ -98,6 +100,7 @@ export const ProviderButton = () => { await refreshRules() await refreshRuleProviders() + void syncRuntimeProviders() showNotice.success('rules.feedback.notifications.provider.allUpdated') } catch (err) { diff --git a/src/services/cmds.ts b/src/services/cmds.ts index d26e48e62..efea521b4 100644 --- a/src/services/cmds.ts +++ b/src/services/cmds.ts @@ -251,6 +251,12 @@ export async function openLogsDir() { return invoke('open_logs_dir').catch((err) => showNotice.error(err)) } +export async function syncRuntimeProviders() { + return invoke('sync_runtime_providers').catch((err) => { + console.warn('failed to queue the provider cache sync', err) + }) +} + export async function cmdTestDelay(url: string) { return invoke('test_delay', { url }) }