From f639b566b898f0f39fd617b7e05df23e4e302485 Mon Sep 17 00:00:00 2001 From: RomainROCH Date: Wed, 14 Oct 2026 23:19:38 +0200 Subject: [PATCH] fix: synchronize status pipe shutdown --- src/ipc.rs | 235 +++++++++++++++++++-------- tests/status_connection_budget.rs | 260 +++++++++++++++++++++++++++++- 2 files changed, 429 insertions(+), 66 deletions(-) diff --git a/src/ipc.rs b/src/ipc.rs index 2337b46..255a94a 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -17,7 +17,7 @@ use std::os::windows::io::{AsRawHandle, FromRawHandle, RawHandle}; use std::sync::{ atomic::{AtomicBool, Ordering}, mpsc::{self, Sender}, - Arc, + Arc, Mutex, MutexGuard, }; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; @@ -59,6 +59,97 @@ const INITIAL_PIPE_CLAIM_DELAY: Duration = Duration::from_millis(100); const CONTROL_PIPE_SECURITY_SDDL_PREFIX: &str = "D:P(A;;GA;;;SY)(A;;GA;;;BA)(A;;GRGW;;;"; const CONTROL_PIPE_SECURITY_SDDL_SUFFIX: &str = ")S:(ML;;NW;;;ME)"; +// The regression fixture installs this only for its process-unique status pipe. +// No hook state or synchronization is compiled into the application. +#[cfg(test)] +#[allow(dead_code)] +pub(crate) mod status_shutdown_test { + use std::sync::{mpsc, Mutex}; + use std::time::Duration; + + #[derive(Debug)] + pub(crate) enum Event { + GateContended, + Wake(Option), + } + + struct Pause { + reached: mpsc::Sender<()>, + resume: mpsc::Receiver<()>, + } + + struct Hook { + endpoint: String, + pause: Option, + wake: Option>, + gate_reported: bool, + } + + static HOOK: Mutex> = Mutex::new(None); + + pub(crate) fn install( + endpoint: String, + ) -> (mpsc::Receiver<()>, mpsc::Sender<()>, mpsc::Receiver) { + let (reached_tx, reached_rx) = mpsc::channel(); + let (resume_tx, resume_rx) = mpsc::channel(); + let (wake_tx, wake_rx) = mpsc::channel(); + let mut hook = HOOK.lock().expect("status regression hook lock"); + assert!(hook.is_none(), "status regression hook already installed"); + *hook = Some(Hook { + endpoint, + pause: Some(Pause { + reached: reached_tx, + resume: resume_rx, + }), + wake: Some(wake_tx), + gate_reported: false, + }); + (reached_rx, resume_tx, wake_rx) + } + + pub(super) fn pause_replacement(endpoint: &str) { + let pause = { + let mut hook = HOOK.lock().expect("status regression hook lock"); + hook.as_mut() + .filter(|hook| hook.endpoint == endpoint) + .and_then(|hook| hook.pause.take()) + }; + if let Some(pause) = pause { + pause.reached.send(()).expect("status regression paused"); + pause + .resume + .recv_timeout(Duration::from_secs(5)) + .expect("status regression resume timed out"); + } + } + + pub(super) fn record_gate_contention(endpoint: &str) { + let mut hook = HOOK.lock().expect("status regression hook lock"); + if let Some(current) = hook.as_mut().filter(|hook| hook.endpoint == endpoint) { + if !current.gate_reported { + if let Some(events) = current.wake.as_ref() { + let _ = events.send(Event::GateContended); + } + current.gate_reported = true; + } + } + } + + pub(super) fn record_wake(endpoint: &str, result: &std::io::Result<()>) { + let mut hook = HOOK.lock().expect("status regression hook lock"); + if let Some(current) = hook.as_mut().filter(|hook| hook.endpoint == endpoint) { + if let Some(wake) = current.wake.take() { + let _ = wake.send(Event::Wake( + result.as_ref().err().and_then(|error| error.raw_os_error()), + )); + } + if current.pause.is_none() && current.wake.is_none() { + *hook = None; + } + } + } +} + /// Wakes the UI thread when the IPC server has queued a new request. pub(crate) const IPC_WAKE_MESSAGE: u32 = WM_APP + 1; @@ -118,7 +209,35 @@ pub(crate) enum ClientError { pub(crate) struct IpcServer { control_join_handle: Option>, status_join_handle: Option>, - status_shutdown_requested: Arc, + status_shutdown_requested: Arc, +} + +#[derive(Default)] +struct StatusShutdown { + requested: AtomicBool, + accept_gate: Mutex<()>, +} + +impl StatusShutdown { + fn lock_accept(&self) -> MutexGuard<'_, ()> { + #[cfg(test)] + if matches!( + self.accept_gate.try_lock(), + Err(std::sync::TryLockError::WouldBlock) + ) { + if let Ok(endpoint) = current_status_pipe_path() { + status_shutdown_test::record_gate_contention(&endpoint); + } + } + self.accept_gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn request(&self) { + let _accept = self.lock_accept(); + self.requested.store(true, Ordering::Release); + } } impl IpcServer { @@ -129,7 +248,7 @@ impl IpcServer { ) -> Result { let control_pipe_path = current_control_pipe_path()?; let status_pipe_path = current_status_pipe_path()?; - let status_shutdown_requested = Arc::new(AtomicBool::new(false)); + let status_shutdown_requested = Arc::new(StatusShutdown::default()); let (startup_tx, startup_rx) = mpsc::channel(); let control_request_tx = request_tx.clone(); let mut control_join_handle = Some( @@ -193,8 +312,7 @@ impl IpcServer { /// Stops accepting new status-pipe clients before the main runtime begins shutdown work. pub(crate) fn begin_shutdown(&self) { - self.status_shutdown_requested - .store(true, Ordering::Release); + self.status_shutdown_requested.request(); let _ = wake_status_server(); } @@ -572,7 +690,7 @@ fn run_server_loop( fn run_status_server_loop( request_tx: Sender, ui_thread_id: u32, - shutdown_requested: Arc, + shutdown_requested: Arc, startup_tx: Sender, ) { let status_pipe_path = match current_status_pipe_path() { @@ -611,44 +729,41 @@ fn run_status_server_loop( let mut initial_pipe = Some(initial_pipe); loop { - if shutdown_requested.load(Ordering::Acquire) { - break; - } + // Publish the next instance before shutdown can set its flag and wake it. + // The synchronous connection wait must stay outside this short gate. + let pipe = { + let _accept = shutdown_requested.lock_accept(); + if shutdown_requested.requested.load(Ordering::Acquire) { + break; + } - let mut pipe = match initial_pipe.take() { - Some(pipe) => match connect_pipe_client(pipe, &status_pipe_path) { - Ok(pipe) => pipe, - Err(error) => { - warn!("Wardoff stopped its status server after an IPC error: {error}"); - logger::log_event( - "status_server_stopped", - EventSource::Ipc, - format!( - "Wardoff stopped its read-only status pipe after an IPC error: {error}" - ), - false, - ); - break; - } - }, - None => match accept_status_client(&status_pipe_path) { - Ok(pipe) => pipe, - Err(error) => { - warn!("Wardoff stopped its status server after an IPC error: {error}"); - logger::log_event( - "status_server_stopped", - EventSource::Ipc, - format!( - "Wardoff stopped its read-only status pipe after an IPC error: {error}" - ), - false, - ); - break; - } - }, + #[cfg(test)] + if initial_pipe.is_none() { + status_shutdown_test::pause_replacement(&status_pipe_path); + } + + match initial_pipe.take() { + Some(pipe) => Ok(pipe), + None => create_status_pipe_instance(&status_pipe_path).map_err(|(error, _)| error), + } + }; + let mut pipe = match pipe.and_then(|pipe| connect_pipe_client(pipe, &status_pipe_path)) { + Ok(pipe) => pipe, + Err(error) => { + warn!("Wardoff stopped its status server after an IPC error: {error}"); + logger::log_event( + "status_server_stopped", + EventSource::Ipc, + format!( + "Wardoff stopped its read-only status pipe after an IPC error: {error}" + ), + false, + ); + break; + } }; - if shutdown_requested.load(Ordering::Acquire) { + if shutdown_requested.requested.load(Ordering::Acquire) { break; } @@ -720,19 +835,13 @@ fn accept_control_client( connect_pipe_client(pipe, control_pipe_path) } -fn accept_status_client(status_pipe_path: &str) -> Result { - let pipe = create_status_pipe_instance(status_pipe_path).map_err(|(error, _)| error)?; - connect_pipe_client(pipe, status_pipe_path) -} - fn wait_for_server_startup( startup_rx: &mpsc::Receiver, control_join_handle: &mut Option>, status_join_handle: &mut Option>, - status_shutdown_requested: &Arc, + status_shutdown_requested: &Arc, ) -> Result<(), String> { let mut control_ready = false; - let mut status_ready = false; for _ in 0..2 { let startup_result = match startup_rx.recv() { @@ -743,7 +852,6 @@ fn wait_for_server_startup( status_join_handle, status_shutdown_requested, control_ready, - status_ready, ); return Err( "Wardoff could not confirm that both named-pipe servers claimed their initial instances." @@ -761,24 +869,21 @@ fn wait_for_server_startup( status_join_handle, status_shutdown_requested, control_ready, - status_ready, ); return Err(error); } }, - ServerStartupKind::Status => match startup_result.result { - Ok(()) => status_ready = true, - Err(error) => { + ServerStartupKind::Status => { + if let Err(error) = startup_result.result { cleanup_failed_server_startup( control_join_handle, status_join_handle, status_shutdown_requested, control_ready, - status_ready, ); return Err(error); } - }, + } } } @@ -788,15 +893,12 @@ fn wait_for_server_startup( fn cleanup_failed_server_startup( control_join_handle: &mut Option>, status_join_handle: &mut Option>, - status_shutdown_requested: &Arc, + status_shutdown_requested: &Arc, control_ready: bool, - status_ready: bool, ) { - status_shutdown_requested.store(true, Ordering::Release); - - if status_ready { - let _ = wake_status_server(); - } + status_shutdown_requested.request(); + // The status worker can already be waiting while its startup ACK is queued. + let _ = wake_status_server(); if control_ready { let _ = send_request(&IpcRequest::ShutdownServer); } @@ -1275,10 +1377,13 @@ fn current_process_user_sid_string(control_pipe_path: &str) -> Result std::io::Result<()> { let status_pipe_path = current_status_pipe_path().map_err(std::io::Error::other)?; - OpenOptions::new() + let result = OpenOptions::new() .read(true) - .open(status_pipe_path) - .map(|_| ()) + .open(&status_pipe_path) + .map(|_| ()); + #[cfg(test)] + status_shutdown_test::record_wake(&status_pipe_path, &result); + result } fn wake_ui_thread(ui_thread_id: u32) -> Result<(), String> { diff --git a/tests/status_connection_budget.rs b/tests/status_connection_budget.rs index a89a2d6..e9cdd81 100644 --- a/tests/status_connection_budget.rs +++ b/tests/status_connection_budget.rs @@ -4,7 +4,10 @@ #![allow(dead_code)] use std::fs::{File, OpenOptions}; +use std::io::Write; use std::os::windows::io::{FromRawHandle, RawHandle}; +use std::path::PathBuf; +use std::process::{Child, Command, ExitStatus, Stdio}; use std::sync::{ atomic::{AtomicBool, AtomicUsize, Ordering}, mpsc, Arc, @@ -12,8 +15,9 @@ use std::sync::{ use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; use windows::core::PCWSTR; -use windows::Win32::Foundation::{GetLastError, ERROR_PIPE_CONNECTED}; +use windows::Win32::Foundation::{GetLastError, ERROR_FILE_NOT_FOUND, ERROR_PIPE_CONNECTED}; use windows::Win32::Storage::FileSystem::{FILE_FLAG_FIRST_PIPE_INSTANCE, PIPE_ACCESS_OUTBOUND}; +use windows::Win32::System::Com::CoCreateGuid; use windows::Win32::System::Pipes::{ ConnectNamedPipe, CreateNamedPipeW, WaitNamedPipeW, PIPE_READMODE_MESSAGE, PIPE_REJECT_REMOTE_CLIENTS, PIPE_TYPE_MESSAGE, PIPE_WAIT, @@ -384,3 +388,257 @@ fn isolated_status_connections_share_a_budget_and_preserve_fallback_rules() { ); } } + +const SHUTDOWN_CHILD: &str = "WARDOFF_STATUS_SHUTDOWN_CHILD"; +const SHUTDOWN_RECEIPT: &str = "WARDOFF_STATUS_SHUTDOWN_RECEIPT"; +const SHUTDOWN_TEST: &str = "isolated_status_shutdown_does_not_lose_the_replacement_wakeup"; +const SHUTDOWN_CHILD_TIMEOUT: Duration = Duration::from_secs(15); +const SHUTDOWN_CLEANUP_TIMEOUT: Duration = Duration::from_secs(5); + +fn shutdown_stage(stage: &str) { + let line = format!( + "[status-shutdown pid={}] stage={stage}\n", + std::process::id() + ); + let _ = std::io::stderr().lock().write_all(line.as_bytes()); +} + +struct ShutdownReceipt(PathBuf); + +impl ShutdownReceipt { + fn reserve() -> Self { + let guid = unsafe { CoCreateGuid() }.expect("status shutdown receipt GUID"); + let path = std::env::temp_dir().join(format!("WardoffStatusShutdown-{guid:?}.receipt")); + OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + .expect("reserve owned status shutdown receipt"); + Self(path) + } +} + +impl Drop for ShutdownReceipt { + fn drop(&mut self) { + if let Err(error) = std::fs::remove_file(&self.0) { + if error.kind() != std::io::ErrorKind::NotFound { + eprintln!("owned status shutdown receipt cleanup failed: {error}"); + } + } + } +} + +struct ShutdownChild(Child); + +impl ShutdownChild { + fn wait_until(&mut self, deadline: Instant) -> std::io::Result> { + loop { + if let Some(status) = self.0.try_wait()? { + return Ok(Some(status)); + } + if Instant::now() >= deadline { + return Ok(None); + } + thread::sleep(Duration::from_millis(10)); + } + } + + fn kill_and_reap(&mut self) -> std::io::Result<()> { + if matches!(self.0.try_wait(), Ok(Some(_))) { + return Ok(()); + } + if let Err(error) = self.0.kill() { + if self.0.try_wait()?.is_some() { + return Ok(()); + } + return Err(error); + } + self.wait_until(Instant::now() + SHUTDOWN_CLEANUP_TIMEOUT)? + .ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::TimedOut, "owned child did not exit") + })?; + Ok(()) + } +} + +impl Drop for ShutdownChild { + fn drop(&mut self) { + if let Err(error) = self.kill_and_reap() { + eprintln!("owned status shutdown child cleanup failed: {error}"); + } + } +} + +fn assert_shutdown_endpoints_removed(pid: u32) { + for kind in ["Control", "Status", "BusyStatus"] { + let endpoint = format!(r"\\.\pipe\WardoffStatusBudget-{kind}-{pid}"); + match OpenOptions::new().read(true).open(endpoint) { + Err(error) => assert_eq!(error.raw_os_error(), Some(ERROR_FILE_NOT_FOUND.0 as i32)), + Ok(file) => { + drop(file); + panic!("owned {kind} endpoint survived status shutdown"); + } + } + } +} + +fn exercise_status_shutdown_replacement() { + let receipt = + PathBuf::from(std::env::var_os(SHUTDOWN_RECEIPT).expect("owned shutdown receipt")); + assert_eq!(receipt.parent(), Some(std::env::temp_dir().as_path())); + assert!(receipt + .file_name() + .expect("receipt name") + .to_string_lossy() + .starts_with("WardoffStatusShutdown-")); + let metadata = std::fs::metadata(&receipt).expect("reserved shutdown receipt"); + assert!(metadata.is_file()); + assert_eq!(metadata.len(), 0); + + let status_endpoint = session_scope::path("Status"); + let (paused, resume, wake) = ipc::status_shutdown_test::install(status_endpoint.clone()); + let mut runtime = TestRuntime::start(); + assert_allow(ipc::read_status(None)); + paused + .recv_timeout(SHUTDOWN_CLEANUP_TIMEOUT) + .expect("replacement worker paused after false shutdown check"); + shutdown_stage("replacement-paused-after-false"); + match OpenOptions::new().read(true).open(&status_endpoint) { + Err(error) => assert_eq!(error.raw_os_error(), Some(ERROR_FILE_NOT_FOUND.0 as i32)), + Ok(file) => { + drop(file); + panic!("replacement endpoint existed while its worker was paused"); + } + } + shutdown_stage("replacement-endpoint-absent"); + + let (stopped_tx, stopped_rx) = mpsc::channel(); + let stopper = thread::spawn(move || { + runtime.stop(); + stopped_tx.send(()).expect("report joined inert runtime"); + }); + // Observe actual mutex contention, or the old implementation's missed wake. + // No scheduling delay is used to choose when the worker resumes. + let early_wake = match wake + .recv_timeout(SHUTDOWN_CLEANUP_TIMEOUT) + .expect("real shutdown reached its gate or wake") + { + ipc::status_shutdown_test::Event::Wake(Some(code)) + if code == ERROR_FILE_NOT_FOUND.0 as i32 => + { + shutdown_stage("real-shutdown-wake-file-not-found"); + Some(Some(code)) + } + ipc::status_shutdown_test::Event::GateContended => { + shutdown_stage("shutdown-gate-contention-confirmed"); + None + } + _ => panic!("shutdown attempted an unexpected wake before pipe publication"), + }; + resume.send(()).expect("resume replacement worker"); + shutdown_stage("replacement-resumed"); + let wake_result = match early_wake { + Some(result) => result, + None => match wake + .recv_timeout(SHUTDOWN_CLEANUP_TIMEOUT) + .expect("real shutdown wake after pipe publication") + { + ipc::status_shutdown_test::Event::Wake(result) => result, + _ => panic!("shutdown wake was not observed after pipe publication"), + }, + }; + if wake_result.is_none() { + shutdown_stage("real-shutdown-wake-connected"); + } + + let stopped_without_rescue = match stopped_rx.recv_timeout(SHUTDOWN_CLEANUP_TIMEOUT) { + Ok(()) => true, + Err(mpsc::RecvTimeoutError::Timeout) => false, + Err(mpsc::RecvTimeoutError::Disconnected) => panic!("inert shutdown thread failed"), + }; + if !stopped_without_rescue { + wait_for_available("Status"); + shutdown_stage("shutdown-incomplete-replacement-available"); + // Connect only our child's endpoint. It releases the synchronous accept; + // the existing post-accept shutdown check then closes it without a reply. + let rescue = OpenOptions::new() + .read(true) + .open(&status_endpoint) + .expect("rescue owned status accept"); + drop(rescue); + shutdown_stage("owned-status-rescue-connected"); + stopped_rx + .recv_timeout(SHUTDOWN_CLEANUP_TIMEOUT) + .expect("inert shutdown completed after rescue"); + } + stopper.join().expect("inert shutdown thread joined"); + shutdown_stage("ipc-and-ui-joined"); + assert_shutdown_endpoints_removed(std::process::id()); + shutdown_stage("owned-endpoints-removed"); + OpenOptions::new() + .write(true) + .open(receipt) + .expect("open owned shutdown receipt") + .write_all(b"cleanup-complete\n") + .expect("record verified shutdown cleanup"); + shutdown_stage("cleanup-receipt-written"); + assert_eq!(wake_result, None, "REGRESSION_MISSED_STATUS_WAKEUP: shutdown attempted its wake while the replacement pipe was absent"); + assert!(stopped_without_rescue, "REGRESSION_MISSED_STATUS_WAKEUP: shutdown required an extra connection after its wake missed the replacement instance"); +} + +#[test] +fn isolated_status_shutdown_does_not_lose_the_replacement_wakeup() { + if let Some(marker) = std::env::var_os(SHUTDOWN_CHILD) { + assert_eq!(marker.to_str(), Some("1"), "invalid owned child marker"); + assert_eq!( + std::env::args().skip(1).collect::>(), + ["--exact", SHUTDOWN_TEST, "--nocapture", "--test-threads=1"], + "owned child requires the exact single-test filter" + ); + exercise_status_shutdown_replacement(); + return; + } + assert!( + std::env::var_os(SHUTDOWN_RECEIPT).is_none(), + "shutdown receipt without owned child marker" + ); + let receipt = ShutdownReceipt::reserve(); + let mut child = ShutdownChild( + Command::new(std::env::current_exe().expect("current test executable")) + .args(["--exact", SHUTDOWN_TEST, "--nocapture", "--test-threads=1"]) + .env(SHUTDOWN_CHILD, "1") + .env(SHUTDOWN_RECEIPT, &receipt.0) + .stdin(Stdio::null()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .expect("spawn owned inert status shutdown child"), + ); + let pid = child.0.id(); + eprintln!("[status-shutdown] owned-child={pid} watchdog={SHUTDOWN_CHILD_TIMEOUT:?}"); + let Some(status) = child + .wait_until(Instant::now() + SHUTDOWN_CHILD_TIMEOUT) + .expect("poll owned shutdown child") + else { + child + .kill_and_reap() + .expect("kill and reap stalled owned shutdown child"); + assert_shutdown_endpoints_removed(pid); + panic!("owned status shutdown child exceeded its watchdog"); + }; + eprintln!("[status-shutdown] owned-child={pid} reaped exit={status}"); + assert_shutdown_endpoints_removed(pid); + shutdown_stage("parent-owned-endpoints-removed"); + assert_eq!( + std::fs::read(&receipt.0).expect("read shutdown cleanup receipt"), + b"cleanup-complete\n", + "child did not verify shutdown cleanup" + ); + std::fs::remove_file(&receipt.0).expect("remove owned shutdown receipt"); + shutdown_stage("parent-owned-receipt-removed"); + // A failed child is always a failed test, even after its rescue and cleanup. + assert!( + status.success(), + "status shutdown regression child failed: {status}" + ); +}