diff --git a/src-tauri/src/core_runtime.rs b/src-tauri/src/core_runtime.rs index 2c0e634..91feb31 100644 --- a/src-tauri/src/core_runtime.rs +++ b/src-tauri/src/core_runtime.rs @@ -2,6 +2,51 @@ use super::*; pub(crate) static CORE_OPERATION_LOCK: Mutex<()> = Mutex::new(()); +pub(crate) fn lock_core_operation( + process_state: &CoreProcessState, +) -> Result, String> { + let guard = CORE_OPERATION_LOCK + .try_lock() + .map_err(|_| "内核正在执行其他操作,请稍后重试".to_string())?; + process_state.ensure_active()?; + Ok(guard) +} + +// Own the guard from spawn, including while the management port is starting. +pub(crate) struct CoreChild { + child: Child, + #[cfg(windows)] + job: isize, +} + +impl std::ops::Deref for CoreChild { + type Target = Child; + + fn deref(&self) -> &Child { + &self.child + } +} + +impl std::ops::DerefMut for CoreChild { + fn deref_mut(&mut self) -> &mut Child { + &mut self.child + } +} + +impl Drop for CoreChild { + fn drop(&mut self) { + #[cfg(windows)] + { + close_windows_handle(self.job); + let _ = self.child.wait(); + } + #[cfg(not(windows))] + if !matches!(self.child.try_wait(), Ok(Some(_))) { + let _ = terminate_child(&mut self.child); + } + } +} + #[cfg(any(target_os = "linux", test))] struct CoreSpawnRequest { command: Command, @@ -30,19 +75,36 @@ static CORE_PROCESS_SPAWNER: LazyLock Result { +pub(crate) fn spawn_core_child(command: Command) -> Result { #[cfg(target_os = "linux")] - { - spawn_core_child_on_lifetime_thread(command) - } + let child = spawn_core_child_on_lifetime_thread(command)?; #[cfg(not(target_os = "linux"))] - { + let child = { let mut command = command; command .spawn() - .map_err(|error| format!("启动 CPA 内核失败: {error}")) - } + .map_err(|error| format!("启动 CPA 内核失败: {error}"))? + }; + + #[cfg(windows)] + let job = match attach_child_to_windows_job(&child) { + Ok(job) => job, + Err(error) => { + let mut child = child; + return match terminate_child(&mut child) { + Ok(()) => Err(error), + Err(cleanup_error) => Err(format!( + "{error};清理未托管的内核进程也失败: {cleanup_error}" + )), + }; + } + }; + Ok(CoreChild { + child, + #[cfg(windows)] + job, + }) } #[cfg(any(target_os = "linux", test))] @@ -63,9 +125,7 @@ async fn run_core_command( operation: fn(&CoreProcessState, &GuiConfigState) -> Result, ) -> Result { tauri::async_runtime::spawn_blocking(move || { - let _guard = CORE_OPERATION_LOCK - .try_lock() - .map_err(|_| "内核正在执行其他操作,请稍后重试".to_string())?; + let _guard = lock_core_operation(app.state::().inner())?; let status = operation( app.state::().inner(), app.state::().inner(), @@ -120,9 +180,7 @@ pub(crate) async fn install_bundled_core( window: tauri::Window, ) -> Result { tauri::async_runtime::spawn_blocking(move || { - let _guard = CORE_OPERATION_LOCK - .try_lock() - .map_err(|_| "内核正在执行其他操作,请稍后重试".to_string())?; + let _guard = lock_core_operation(app.state::().inner())?; let state = app.state::(); let process_state = app.state::(); let gui_config_state = app.state::(); @@ -190,9 +248,7 @@ pub(crate) async fn install_core_version( version: Option, ) -> Result { tauri::async_runtime::spawn_blocking(move || { - let _guard = CORE_OPERATION_LOCK - .try_lock() - .map_err(|_| "内核正在执行其他操作,请稍后重试".to_string())?; + let _guard = lock_core_operation(app.state::().inner())?; let state = app.state::(); let process_state = app.state::(); let gui_config_state = app.state::(); @@ -256,10 +312,21 @@ fn pause_core_for_install( process_state: &CoreProcessState, config: &GuiConfigFile, ) -> Result { - let was_running = current_core_status(Some(process_state), Some(config.port))?.running; + let was_running = pause_core_process_for_install(process_state)?; + if was_running { + emit_current_core_status(app, process_state, config.port); + } + Ok(was_running) +} + +pub(crate) fn pause_core_process_for_install( + process_state: &CoreProcessState, +) -> Result { + process_state.ensure_active()?; + // A process can hold files even when its HTTP port is not responding. + let was_running = current_core_status(Some(process_state), None)?.running; if was_running { stop_core_process_inner(process_state)?; - emit_current_core_status(app, process_state, config.port); } Ok(was_running) } @@ -271,7 +338,7 @@ fn restore_core_after_install( was_running: bool, install_result: Result, ) -> Result { - let restart_result = if was_running { + let restart_result = if was_running && !process_state.is_shutting_down() { start_core_process_inner(process_state, config) } else { Ok(()) @@ -344,7 +411,9 @@ pub(crate) fn restart_core_process_with_state( gui_config_state: &GuiConfigState, ) -> Result { let config = gui_config_state.snapshot()?; - let _ = stop_core_process_inner(process_state); + if current_core_status(Some(process_state), None)?.running { + stop_core_process_inner(process_state)?; + } start_core_process_inner(process_state, &config)?; if let Err(error) = gui_config_state.set_run_on_startup(true) { let _ = stop_core_process_inner(process_state); @@ -1202,6 +1271,7 @@ fn core_management_address(listen_host: &str, port: u16) -> Result) -> std::fmt::Result { match self { + Self::ShuttingDown => formatter.write_str("应用正在退出,已取消内核启动"), Self::Exited(status) => write!(formatter, "CPA 内核启动后立即退出: {status}"), Self::Spawn(error) => formatter.write_str(error), Self::StatusCheck(error) => write!(formatter, "检查 CPA 内核启动状态失败: {error}"), @@ -1299,12 +1370,13 @@ impl CoreStartAttemptFailure { } fn start_core_process_once( + process_state: &CoreProcessState, binary_path: &Path, config_path: &str, install_dir: &Path, log_path: &Path, management_address: SocketAddr, -) -> Result { +) -> Result { let mut command = Command::new(binary_path); command .args(["-config", config_path]) @@ -1331,7 +1403,7 @@ fn start_core_process_once( }); } }; - match wait_for_core_management_port(&mut child, management_address) { + match wait_for_core_management_port(&mut child, management_address, process_state) { Ok(()) => Ok(child), Err(failure) => { if !failure.child_has_exited() { @@ -1346,6 +1418,7 @@ pub(crate) fn start_core_process_inner( process_state: &CoreProcessState, gui_config: &GuiConfigFile, ) -> Result<(), String> { + process_state.ensure_active()?; let install_dir = core_install_dir()?; if !gui_config.auth_dir.trim().is_empty() { let auth_dir = auth_dir_path_for_core(&gui_config.auth_dir, &install_dir); @@ -1375,6 +1448,7 @@ pub(crate) fn start_core_process_inner( let log_path = core_start_log_path(&install_dir, &gui_config.auth_dir); let start_once = || { start_core_process_once( + process_state, &binary_path, &config_path, &install_dir, @@ -1416,9 +1490,13 @@ pub(crate) fn start_core_process_inner( pub(crate) fn wait_for_core_management_port( child: &mut Child, address: SocketAddr, + process_state: &CoreProcessState, ) -> Result<(), CoreStartupFailure> { let deadline = Instant::now() + Duration::from_secs(10); loop { + if process_state.is_shutting_down() { + return Err(CoreStartupFailure::ShuttingDown); + } if let Some(status) = child.try_wait().map_err(CoreStartupFailure::StatusCheck)? { return Err(CoreStartupFailure::Exited(status)); } @@ -1500,7 +1578,6 @@ pub(crate) fn stop_core_process_inner(process_state: &CoreProcessState) -> Resul if let Err(error) = terminate_child(&mut child) { errors.push(error); } - process_state.clear_lifetime_guard(); } let mut process_ids = process_state @@ -2113,8 +2190,14 @@ pub(crate) fn shutdown_managed_core( .ok() .and_then(|install_dir| find_core_binary(&install_dir)) .is_some_and(|binary_path| is_core_running(&binary_path)); - let _ = gui_config_state.set_run_on_startup(was_running); - let _ = stop_core_process_inner(process_state); + if was_running { + if let Err(error) = stop_core_process_inner(process_state) { + eprintln!("退出时关闭 CPA 内核失败: {error}"); + } + } + if let Err(error) = gui_config_state.set_run_on_startup(was_running) { + eprintln!("保存退出前的内核状态失败: {error}"); + } } #[cfg(windows)] @@ -2173,6 +2256,13 @@ pub(crate) fn close_windows_handle(handle: isize) { } pub(crate) fn terminate_child(child: &mut Child) -> Result<(), String> { + if child + .try_wait() + .map_err(|error| format!("检查 CPA 内核进程状态失败: {error}"))? + .is_some() + { + return Ok(()); + } #[cfg(windows)] { child @@ -2211,10 +2301,31 @@ pub(crate) fn terminate_child(child: &mut Child) -> Result<(), String> { pub(crate) fn terminate_process_id(process_id: u32) -> Result<(), String> { #[cfg(windows)] { - let process_id = process_id.to_string(); + use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; + use windows_sys::Win32::{ + Foundation::{ERROR_INVALID_PARAMETER, WAIT_OBJECT_0, WAIT_TIMEOUT}, + System::Threading::{OpenProcess, WaitForSingleObject, PROCESS_SYNCHRONIZE}, + }; + + // Keep a handle to this process until it is signalled. taskkill exiting + // successfully only means termination was requested, not that file + // handles have been released. + let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE, 0, process_id) }; + if handle.is_null() { + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) { + return Ok(()); + } + return Err(format!("打开 CPA 内核进程失败: PID {process_id}: {error}")); + } + let handle = unsafe { OwnedHandle::from_raw_handle(handle) }; + let raw_handle = handle.as_raw_handle(); + if unsafe { WaitForSingleObject(raw_handle, 0) } == WAIT_OBJECT_0 { + return Ok(()); + } let mut command = Command::new("taskkill"); command - .args(["/PID", &process_id, "/T", "/F"]) + .args(["/PID", &process_id.to_string(), "/T", "/F"]) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()); @@ -2222,11 +2333,15 @@ pub(crate) fn terminate_process_id(process_id: u32) -> Result<(), String> { let status = command .status() .map_err(|err| format!("关闭 CPA 内核进程失败: {err}"))?; - - if status.success() { - Ok(()) - } else { - Err(format!("关闭 CPA 内核进程失败: PID {process_id}")) + let wait = + unsafe { WaitForSingleObject(raw_handle, if status.success() { 10_000 } else { 0 }) }; + match wait { + WAIT_OBJECT_0 => Ok(()), + WAIT_TIMEOUT => Err(format!("CPA 内核进程未退出: PID {process_id}")), + _ => Err(format!( + "等待 CPA 内核进程退出失败: PID {process_id}: {}", + io::Error::last_os_error() + )), } } diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index cb0e353..a8b7ca1 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -303,11 +303,11 @@ struct AppUpdateInner { #[derive(Default)] struct CoreProcessState { - child: Mutex>, + child: Mutex>, adopted_processes: Mutex>, starting: AtomicBool, - #[cfg(windows)] - job: Mutex>, + shutting_down: AtomicBool, + shutdown_complete: AtomicBool, } #[derive(Default)] @@ -1739,8 +1739,20 @@ impl CoreProcessState { child: Mutex::new(None), adopted_processes: Mutex::new(Vec::new()), starting: AtomicBool::new(starting), - #[cfg(windows)] - job: Mutex::new(None), + shutting_down: AtomicBool::new(false), + shutdown_complete: AtomicBool::new(false), + } + } + + fn is_shutting_down(&self) -> bool { + self.shutting_down.load(Ordering::Acquire) + } + + fn ensure_active(&self) -> Result<(), String> { + if self.is_shutting_down() { + Err("应用正在退出,已取消内核操作".to_string()) + } else { + Ok(()) } } @@ -1760,8 +1772,6 @@ impl CoreProcessState { } *child = None; - drop(child); - self.clear_lifetime_guard(); } } @@ -1778,16 +1788,7 @@ impl CoreProcessState { }) } - fn clear_lifetime_guard(&self) { - #[cfg(windows)] - if let Ok(mut job) = self.job.lock() { - if let Some(handle) = job.take() { - close_windows_handle(handle); - } - } - } - - fn take_child(&self) -> Option { + fn take_child(&self) -> Option { self.child.lock().ok().and_then(|mut child| child.take()) } @@ -1823,55 +1824,15 @@ impl CoreProcessState { .unwrap_or_default() } - fn store_child(&self, child: Child) -> Result { + fn store_child(&self, child: CoreChild) -> Result { + self.ensure_active()?; let pid = child.id(); self.clear_adopted_processes()?; - - #[cfg(windows)] - { - let job = match attach_child_to_windows_job(&child) { - Ok(job) => job, - Err(error) => { - let mut child = child; - let cleanup_error = terminate_child(&mut child).err(); - return Err(match cleanup_error { - Some(cleanup_error) => { - format!("{error};清理未托管的内核进程也失败: {cleanup_error}") - } - None => error, - }); - } - }; - let Ok(mut managed_child) = self.child.lock() else { - close_windows_handle(job); - return Err("内核进程状态锁已损坏".to_string()); - }; - let Ok(mut managed_job) = self.job.lock() else { - close_windows_handle(job); - return Err("内核进程作业状态锁已损坏".to_string()); - }; - *managed_child = Some(child); - *managed_job = Some(job); - } - - #[cfg(not(windows))] - { - let mut managed_child = match self.child.lock() { - Ok(managed_child) => managed_child, - Err(_) => { - let mut child = child; - let cleanup_error = terminate_child(&mut child).err(); - return Err(match cleanup_error { - Some(cleanup_error) => format!( - "内核进程状态锁已损坏;清理未托管的内核进程也失败: {cleanup_error}" - ), - None => "内核进程状态锁已损坏".to_string(), - }); - } - }; - *managed_child = Some(child); - } - + let mut managed_child = self + .child + .lock() + .map_err(|_| "内核进程状态锁已损坏".to_string())?; + *managed_child = Some(child); Ok(pid) } } @@ -2448,6 +2409,10 @@ fn main() { }; let gui_config_state = core_app.state::(); let process_state = core_app.state::(); + if process_state.ensure_active().is_err() { + process_state.set_starting(false); + return; + } let Ok(config) = gui_config_state.snapshot() else { process_state.set_starting(false); return; @@ -2623,10 +2588,34 @@ fn main() { has_visible_windows: false, .. } => show_main_window(app_handle), - tauri::RunEvent::ExitRequested { .. } => { + tauri::RunEvent::ExitRequested { api, code, .. } => { + let process_state = app_handle.state::(); + if process_state.shutdown_complete.load(Ordering::Acquire) { + return; + } + api.prevent_exit(); + if process_state.shutting_down.swap(true, Ordering::AcqRel) { + return; + } if let Err(error) = persist_main_window_size(app_handle) { eprintln!("保存主窗口尺寸失败: {error}"); } + app_handle.state::().cancel(); + let app_handle = app_handle.clone(); + // Keep the event loop alive while an in-flight operation finishes: + // tray/status updates from that operation may need the UI thread. + tauri::async_runtime::spawn_blocking(move || { + let _guard = CORE_OPERATION_LOCK + .lock() + .unwrap_or_else(|error| error.into_inner()); + let process_state = app_handle.state::(); + let gui_config_state = app_handle.state::(); + shutdown_managed_core(process_state.inner(), gui_config_state.inner()); + process_state + .shutdown_complete + .store(true, Ordering::Release); + app_handle.exit(code.unwrap_or(0)); + }); } tauri::RunEvent::Exit => { usage::stop_usage_collector(app_handle); @@ -2636,7 +2625,10 @@ fn main() { } let gui_config_state = app_handle.state::(); let process_state = app_handle.state::(); - shutdown_managed_core(process_state.inner(), gui_config_state.inner()); + if !process_state.shutdown_complete.load(Ordering::Acquire) { + process_state.shutting_down.store(true, Ordering::Release); + shutdown_managed_core(process_state.inner(), gui_config_state.inner()); + } } _ => {} }); diff --git a/src-tauri/src/tests/core_runtime.rs b/src-tauri/src/tests/core_runtime.rs index 988b93f..6946b7d 100644 --- a/src-tauri/src/tests/core_runtime.rs +++ b/src-tauri/src/tests/core_runtime.rs @@ -50,7 +50,7 @@ fn core_child_sleep_command() -> Command { command } -fn assert_core_child_survives(mut child: Child) { +fn assert_core_child_survives(child: &mut Child) { thread::sleep(Duration::from_millis(200)); let status = child.try_wait(); let _ = child.kill(); @@ -60,10 +60,10 @@ fn assert_core_child_survives(mut child: Child) { #[test] fn core_child_survives_launcher_thread_exit() { - let child = thread::spawn(|| spawn_core_child(core_child_sleep_command()).unwrap()) + let mut child = thread::spawn(|| spawn_core_child(core_child_sleep_command()).unwrap()) .join() .unwrap(); - assert_core_child_survives(child); + assert_core_child_survives(&mut child); } #[test] @@ -71,25 +71,152 @@ fn core_child_survives_blocking_runtime_shutdown() { let runtime = tokio::runtime::Builder::new_current_thread() .build() .unwrap(); - let child = runtime + let mut child = runtime .block_on(runtime.spawn_blocking(|| spawn_core_child(core_child_sleep_command()).unwrap())) .unwrap(); drop(runtime); - assert_core_child_survives(child); + assert_core_child_survives(&mut child); } #[test] fn core_child_spawner_remains_available_after_spawn_failure() { let missing_binary = agent_test_home("missing-core-spawner-binary").join(core_binary_name()); assert!(spawn_core_child_on_lifetime_thread(Command::new(missing_binary)).is_err()); - let child = + let mut child = thread::spawn(|| spawn_core_child_on_lifetime_thread(core_child_sleep_command()).unwrap()) .join() .unwrap(); - assert_core_child_survives(child); + assert_core_child_survives(&mut child); } -#[cfg(target_os = "linux")] +#[test] +fn exiting_during_core_startup_cancels_the_port_wait() { + let state = std::sync::Arc::new(CoreProcessState::new(true)); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + drop(listener); + let mut child = spawn_core_child(core_child_sleep_command()).unwrap(); + let child_id = child.id(); + let exiting_state = state.clone(); + let exit = thread::spawn(move || { + thread::sleep(Duration::from_millis(100)); + exiting_state.shutting_down.store(true, Ordering::Release); + }); + + let started = Instant::now(); + let result = wait_for_core_management_port(&mut child, address, &state); + drop(child); + exit.join().unwrap(); + assert!(matches!(result, Err(CoreStartupFailure::ShuttingDown))); + assert!(started.elapsed() < Duration::from_secs(2)); + assert!(!is_process_alive(child_id)); +} + +#[test] +fn exiting_rejects_new_core_operations_and_cleans_up_an_in_flight_child() { + let state = CoreProcessState::new(false); + let child = spawn_core_child(core_child_sleep_command()).unwrap(); + let child_id = child.id(); + state.shutting_down.store(true, Ordering::Release); + + assert!(lock_core_operation(&state) + .unwrap_err() + .contains("应用正在退出")); + assert!(state + .store_child(child) + .unwrap_err() + .contains("应用正在退出")); + assert!(!is_process_alive(child_id)); + assert_eq!(state.managed_pid(), None); +} + +#[cfg(windows)] +#[test] +fn core_config_file_lock_helper() { + use std::os::windows::fs::OpenOptionsExt; + use windows_sys::Win32::Storage::FileSystem::{FILE_SHARE_READ, FILE_SHARE_WRITE}; + + let Some(path) = env::var_os("EASYCLIPROXYAPI_CORE_FILE_LOCK_TEST_HELPER") else { + return; + }; + let _file = File::options() + .read(true) + .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE) + .open(path) + .unwrap(); + println!("CORE_FILE_LOCKED"); + io::stdout().flush().unwrap(); + thread::sleep(Duration::from_secs(10)); +} + +#[cfg(windows)] +#[test] +fn updating_stops_an_adopted_core_without_a_port_before_replacing_its_config() { + use std::io::BufRead; + + let root = agent_test_home("core-stop-before-replace"); + let config_path = root.join(CORE_CONFIG_FILE); + let replacement = root.join("new-config.yaml"); + fs::write(&config_path, b"old config").unwrap(); + fs::write(&replacement, b"new config").unwrap(); + let binary_path = env::current_exe().unwrap(); + let mut command = Command::new(&binary_path); + command + .args([ + "--exact", + "tests::core_runtime::core_config_file_lock_helper", + "--nocapture", + ]) + .env("EASYCLIPROXYAPI_CORE_FILE_LOCK_TEST_HELPER", &config_path) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()); + configure_background_command(&mut command); + // Spawn without the GUI's job guard to simulate a previous orphan. + let mut child = command.spawn().unwrap(); + let output = io::BufReader::new(child.stdout.take().unwrap()); + let ready = output + .lines() + .any(|line| line.unwrap() == "CORE_FILE_LOCKED"); + let state = CoreProcessState::new(false); + state + .adopt_process_ids(&binary_path, vec![child.id()]) + .unwrap(); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + drop(listener); + + let status = current_core_status(Some(&state), Some(port)).unwrap(); + let locked_result = copy_core_file_replace(&replacement, &config_path); + let stopped = pause_core_process_for_install(&state); + let exited = child.try_wait().unwrap().is_some(); + let replace_result = copy_core_file_replace(&replacement, &config_path); + let _ = child.kill(); + let _ = child.wait(); + + assert!(ready); + assert!( + !status.running, + "the orphan has no listening management port" + ); + assert!( + locked_result.is_err(), + "the orphan must hold a real Windows file lock" + ); + assert!( + stopped.unwrap(), + "the update must stop the orphan despite the closed port" + ); + assert!( + exited, + "stopping must wait until the process has actually exited" + ); + replace_result.unwrap(); + assert_eq!(fs::read(&config_path).unwrap(), b"new config"); + fs::remove_dir_all(root).unwrap(); +} + +#[cfg(any(target_os = "linux", windows))] #[test] fn core_child_owner_process_helper() { if env::var_os("EASYCLIPROXYAPI_CORE_OWNER_TEST_HELPER").is_none() { @@ -103,7 +230,7 @@ fn core_child_owner_process_helper() { thread::sleep(Duration::from_secs(10)); } -#[cfg(target_os = "linux")] +#[cfg(any(target_os = "linux", windows))] #[test] fn core_child_stops_when_owner_process_is_killed() { use std::io::BufRead; diff --git a/src-tauri/src/tray.rs b/src-tauri/src/tray.rs index edaf68c..1907270 100644 --- a/src-tauri/src/tray.rs +++ b/src-tauri/src/tray.rs @@ -473,6 +473,7 @@ pub(crate) fn run_windows_tray_core_action( let process_state = app_handle.state::(); let gui_config_state = app_handle.state::(); let result = (|| -> Result { + let _guard = lock_core_operation(process_state.inner())?; match action { WindowsTrayCoreAction::Toggle => { let config = gui_config_state.snapshot()?;