refactor service manager locking

src-tauri/src/cmd/service.rs

@@ -1,13 +1,11 @@
 use super::{CmdResult, StringifyErr as _};
 use crate::core::service::{self, SERVICE_MANAGER, ServiceStatus};
-use smartstring::SmartString;
 
 async fn execute_service_operation_sync(status: ServiceStatus, op_type: &str) -> CmdResult {
-    if let Err(e) = SERVICE_MANAGER.lock().await.handle_service_status(&status).await {
-        let emsg = format!("{} Service failed: {}", op_type, e);
-        return Err(SmartString::from(emsg));
-    }
-    Ok(())
+    SERVICE_MANAGER
+        .handle_service_status(status)
+        .await
+        .map_err(|e| format!("{op_type} Service failed: {e}").into())
 }
 
 #[tauri::command]

src-tauri/src/core/manager/lifecycle.rs

@@ -12,7 +12,7 @@ use tauri_plugin_clash_verge_sysinfo;
 
 impl CoreManager {
     pub async fn start_core(&self) -> Result<()> {
-        self.prepare_startup().await?;
+        self.prepare_startup().await;
         defer! {
             self.after_core_process();
         }
@@ -62,18 +62,13 @@ impl CoreManager {
         Ok(())
     }
 
-    async fn prepare_startup(&self) -> Result<()> {
+    async fn prepare_startup(&self) {
         #[cfg(target_os = "windows")]
         self.wait_for_service_if_needed().await;
-
-        let value = SERVICE_MANAGER.lock().await.current();
-        let mode = match value {
+        self.set_running_mode(match SERVICE_MANAGER.current().await {
             ServiceStatus::Ready => RunningMode::Service,
             _ => RunningMode::Sidecar,
-        };
-
-        self.set_running_mode(mode);
-        Ok(())
+        });
     }
 
     fn after_core_process(&self) {
@@ -98,9 +93,7 @@ impl CoreManager {
             .with_max_times(max_times as usize);
 
         let _ = (|| async {
-            let mut manager = SERVICE_MANAGER.lock().await;
-
-            if matches!(manager.current(), ServiceStatus::Ready) {
+            if matches!(SERVICE_MANAGER.current().await, ServiceStatus::Ready) {
                 return Ok(());
             }
 
@@ -110,10 +103,10 @@ impl CoreManager {
                 return Err(anyhow::anyhow!("Service IPC not ready"));
             }
 
-            manager.init().await?;
-            let _ = manager.refresh().await;
+            SERVICE_MANAGER.init().await?;
+            let _ = SERVICE_MANAGER.refresh().await;
 
-            if matches!(manager.current(), ServiceStatus::Ready) {
+            if matches!(SERVICE_MANAGER.current().await, ServiceStatus::Ready) {
                 Ok(())
             } else {
                 Err(anyhow::anyhow!("Service not ready"))

src-tauri/src/core/service.rs

@@ -3,20 +3,24 @@ use crate::{
     core::{logger::Logger, tray::Tray},
     utils::dirs,
 };
-use anyhow::{Context as _, Result, anyhow, bail};
+use anyhow::{Context as _, Result, bail};
 use backon::{ConstantBuilder, Retryable as _};
-use clash_verge_logging::{Type, logging, logging_error};
+use clash_verge_logging::{Type, logging};
 use clash_verge_service_ipc::CoreConfig;
 use compact_str::CompactString;
 use once_cell::sync::Lazy;
+use parking_lot::Mutex;
+use scopeguard::defer;
 use std::{
     borrow::Cow,
     env::current_exe,
+    future::Future,
     path::{Path, PathBuf},
     process::Command as StdCommand,
+    sync::atomic::{AtomicBool, Ordering},
     time::Duration,
 };
-use tokio::sync::Mutex;
+use tokio::sync::Notify;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ServiceStatus {
@@ -29,8 +33,11 @@ pub enum ServiceStatus {
     Unavailable(String),
 }
 
-#[derive(Clone)]
-pub struct ServiceManager(ServiceStatus);
+pub struct ServiceManager {
+    status: Mutex<ServiceStatus>,
+    operation_running: AtomicBool,
+    operation_done: Notify,
+}
 
 #[cfg(target_os = "windows")]
 fn uninstall_service() -> Result<()> {
@@ -379,7 +386,7 @@ pub(super) async fn start_with_existing_service(config_file: &PathBuf) -> Result
 pub(super) async fn run_core_by_service(config_file: &PathBuf) -> Result<()> {
     logging!(info, Type::Service, "正在尝试通过服务启动核心");
 
-    SERVICE_MANAGER.lock().await.refresh().await?;
+    SERVICE_MANAGER.refresh().await?;
 
     logging!(info, Type::Service, "服务已运行且版本匹配，直接使用");
     start_with_existing_service(config_file).await
@@ -435,24 +442,7 @@ pub async fn is_service_available() -> Result<()> {
     Ok(())
 }
 
-pub async fn wait_and_check_service_available(status: &mut ServiceManager) -> Result<()> {
-    wait_for_service_ipc(status, "Waiting for service to be available").await
-}
-
-async fn wait_and_check_service_version(status: &mut ServiceManager) -> Result<()> {
-    wait_and_check_service_available(status).await?;
-
-    if clash_verge_service_ipc::is_reinstall_service_needed().await {
-        logging!(info, Type::Service, "服务版本不匹配，执行重装流程");
-        reinstall_service()?;
-        wait_and_check_service_available(status).await?;
-    }
-
-    Ok(())
-}
-
-async fn wait_for_service_ipc(status: &mut ServiceManager, reason: &str) -> Result<()> {
-    status.0 = ServiceStatus::Unavailable(reason.into());
+async fn wait_for_service_ipc(manager: &ServiceManager) -> Result<()> {
     let config = ServiceManager::config();
 
     let backoff = ConstantBuilder::default()
@@ -460,18 +450,18 @@ async fn wait_for_service_ipc(status: &mut ServiceManager, reason: &str) -> Resu
         .with_max_times(config.max_retries);
 
     let result = (|| async {
-        if Path::new(clash_verge_service_ipc::IPC_PATH).exists() {
-            clash_verge_service_ipc::connect().await?;
-            Ok(())
-        } else {
-            Err(anyhow!("IPC path not ready"))
+        if !is_service_ipc_path_exists() {
+            bail!("IPC path not ready");
         }
+        clash_verge_service_ipc::connect().await.map(drop)
     })
     .retry(backoff)
     .await;
 
     if result.is_ok() {
-        status.0 = ServiceStatus::Ready;
+        manager.set_status(ServiceStatus::Ready);
+    } else {
+        manager.set_status(ServiceStatus::Unavailable("Waiting for service to be available".into()));
     }
 
     result
@@ -482,10 +472,6 @@ pub fn is_service_ipc_path_exists() -> bool {
 }
 
 impl ServiceManager {
-    pub fn default() -> Self {
-        Self(ServiceStatus::Unavailable("Need Checks".into()))
-    }
-
     pub const fn config() -> clash_verge_service_ipc::IpcConfig {
         clash_verge_service_ipc::IpcConfig {
             default_timeout: Duration::from_millis(150),
@@ -494,72 +480,109 @@ impl ServiceManager {
         }
     }
 
-    pub async fn init(&mut self) -> Result<()> {
+    pub async fn init(&self) -> Result<()> {
         if let Err(e) = clash_verge_service_ipc::connect().await {
-            self.0 = ServiceStatus::Unavailable("服务连接失败: {e}".to_string());
+            self.set_status(ServiceStatus::Unavailable("服务连接失败: {e}".to_string()));
             return Err(e);
         }
         Ok(())
     }
 
-    pub fn current(&self) -> ServiceStatus {
-        self.0.clone()
-    }
-
-    pub async fn refresh(&mut self) -> Result<()> {
-        let status = self.check_service_comprehensive().await;
-        self.0 = status.clone();
-        logging_error!(Type::Service, self.handle_service_status(&status).await);
-        Ok(())
-    }
-
-    /// 综合服务状态检查（一次性完成所有检查）
-    pub async fn check_service_comprehensive(&self) -> ServiceStatus {
-        if clash_verge_service_ipc::is_reinstall_service_needed().await {
-            ServiceStatus::NeedsReinstall
-        } else {
-            ServiceStatus::Ready
+    pub async fn current(&self) -> ServiceStatus {
+        loop {
+            let notified = self.operation_done.notified();
+            if !self.operation_running.load(Ordering::Acquire) {
+                let status = self.status.lock().clone();
+                if !self.operation_running.load(Ordering::Acquire) {
+                    return status;
+                }
+            }
+            notified.await;
         }
     }
 
-    /// 根据服务状态执行相应操作
-    pub async fn handle_service_status(&mut self, status: &ServiceStatus) -> Result<()> {
-        match status {
-            ServiceStatus::Ready => {
-                logging!(info, Type::Service, "服务就绪，直接启动");
-                self.0 = ServiceStatus::Ready;
+    fn set_status(&self, status: ServiceStatus) {
+        *self.status.lock() = status;
+    }
+
+    async fn run_operation(&self, operation: impl Future<Output = Result<()>>) -> Result<()> {
+        {
+            if self.operation_running.swap(true, Ordering::AcqRel) {
+                bail!("service operation already running");
             }
+            defer! {
+                self.operation_running.store(false, Ordering::Release);
+                self.operation_done.notify_waiters();
+            }
+
+            operation.await?;
+        }
+
+        Tray::global().update_menu().await
+    }
+
+    pub async fn refresh(&self) -> Result<()> {
+        self.run_operation(async {
+            self.apply_service_status(if clash_verge_service_ipc::is_reinstall_service_needed().await {
+                ServiceStatus::NeedsReinstall
+            } else {
+                ServiceStatus::Ready
+            })
+            .await
+        })
+        .await
+    }
+
+    pub async fn handle_service_status(&self, status: ServiceStatus) -> Result<()> {
+        self.run_operation(self.apply_service_status(status)).await
+    }
+
+    async fn apply_service_status(&self, status: ServiceStatus) -> Result<()> {
+        self.set_status(status.clone());
+        match status {
+            ServiceStatus::Ready => logging!(info, Type::Service, "服务就绪，直接启动"),
             ServiceStatus::NeedsReinstall | ServiceStatus::ReinstallRequired => {
                 logging!(info, Type::Service, "服务需要重装，执行重装流程");
-                reinstall_service()?;
-                wait_and_check_service_available(self).await?;
+                run_service_command(reinstall_service, "reinstall service")?;
+                wait_for_service_ipc(self).await?;
             }
             ServiceStatus::ForceReinstallRequired => {
                 logging!(info, Type::Service, "服务需要强制重装，执行强制重装流程");
-                force_reinstall_service()?;
-                wait_and_check_service_available(self).await?;
+                run_service_command(force_reinstall_service, "force reinstall service")?;
+                wait_for_service_ipc(self).await?;
             }
             ServiceStatus::InstallRequired => {
                 logging!(info, Type::Service, "需要安装服务，执行安装流程");
-                install_service()?;
-                wait_and_check_service_version(self).await?;
+                run_service_command(install_service, "install service")?;
+                wait_for_service_ipc(self).await?;
+                if clash_verge_service_ipc::is_reinstall_service_needed().await {
+                    logging!(info, Type::Service, "服务版本不匹配，执行重装流程");
+                    self.set_status(ServiceStatus::NeedsReinstall);
+                    run_service_command(reinstall_service, "reinstall service")?;
+                    wait_for_service_ipc(self).await?;
+                }
             }
             ServiceStatus::UninstallRequired => {
                 logging!(info, Type::Service, "服务需要卸载，执行卸载流程");
-                uninstall_service()?;
-                self.0 = ServiceStatus::Unavailable("Service Uninstalled".into());
+                run_service_command(uninstall_service, "uninstall service")?;
+                self.set_status(ServiceStatus::Unavailable("Service Uninstalled".into()));
             }
             ServiceStatus::Unavailable(reason) => {
                 logging!(info, Type::Service, "服务不可用: {}，将使用Sidecar模式", reason);
-                self.0 = ServiceStatus::Unavailable(reason.clone());
-                return Err(anyhow::anyhow!("服务不可用: {}", reason));
+                bail!("服务不可用: {}", reason);
             }
         }
 
-        // 防止服务安装成功后，内核未完全启动导致系统托盘无法获取代理节点信息
-        Tray::global().update_menu().await?;
         Ok(())
     }
 }
 
-pub static SERVICE_MANAGER: Lazy<Mutex<ServiceManager>> = Lazy::new(|| Mutex::new(ServiceManager::default()));
+fn run_service_command(operation: impl FnOnce() -> Result<()>, label: &'static str) -> Result<()> {
+    tokio::task::block_in_place(operation).with_context(|| format!("{label} failed"))
+}
+
+pub static SERVICE_MANAGER: Lazy<ServiceManager> = Lazy::new(|| ServiceManager {
+    status: Mutex::new(ServiceStatus::Unavailable("Need Checks".into())),
+    operation_running: AtomicBool::new(false),
+    operation_done: Notify::new(),
+});

src-tauri/src/utils/resolve/mod.rs

@@ -175,14 +175,9 @@ pub(super) async fn init_verge_config() {
 
 pub(super) async fn init_service_manager() {
     clash_verge_service_ipc::set_config(Some(ServiceManager::config())).await;
-    if !is_service_ipc_path_exists() {
-        return;
+    if is_service_ipc_path_exists() && SERVICE_MANAGER.init().await.is_ok() {
+        logging_error!(Type::Setup, SERVICE_MANAGER.refresh().await);
     }
-    let mut manager = SERVICE_MANAGER.lock().await;
-    if manager.init().await.is_ok() {
-        logging_error!(Type::Setup, manager.refresh().await);
-    }
-    drop(manager);
 }
 
 pub(super) async fn init_core_manager() {