diff --git a/Cargo.lock b/Cargo.lock index b6ef0549b..58690e40a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1224,6 +1224,7 @@ dependencies = [ "nix 0.31.2", "ntest", "ouroboros", + "pipe_socket", "rustc-hash", "sha2 0.11.0", "subprocess_test", @@ -1298,7 +1299,6 @@ dependencies = [ "fspy_shared_unix", "libc", "nix 0.31.2", - "wincode", ] [[package]] @@ -1340,20 +1340,11 @@ dependencies = [ name = "fspy_shared" version = "0.0.0" dependencies = [ - "assert2", "bitflags 2.10.0", - "bstr", "bumpalo", "bytemuck", - "ctor", - "fspy_shm", "native_str", - "rustc-hash", - "subprocess_test", - "thiserror 2.0.18", - "tokio", - "tracing", - "uuid", + "pipe_socket", "vite_path", "winapi", "wincode", diff --git a/Cargo.toml b/Cargo.toml index 2a78ecd91..20c82dda0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -78,7 +78,6 @@ fspy_detours_sys = { path = "crates/fspy_detours_sys" } fspy_preload_unix = { path = "crates/fspy_preload_unix", artifact = "cdylib", target = "target" } fspy_preload_windows = { path = "crates/fspy_preload_windows", artifact = "cdylib", target = "target" } fspy_seccomp_unotify = { path = "crates/fspy_seccomp_unotify" } -fspy_shm = { path = "crates/fspy_shm" } fspy_shared = { path = "crates/fspy_shared" } fspy_shared_unix = { path = "crates/fspy_shared_unix" } futures = "0.3.31" diff --git a/crates/fspy/Cargo.toml b/crates/fspy/Cargo.toml index 38b9ffb51..6c9d08b1a 100644 --- a/crates/fspy/Cargo.toml +++ b/crates/fspy/Cargo.toml @@ -15,6 +15,7 @@ fspy_shared = { workspace = true } futures-util = { workspace = true } libc = { workspace = true } ouroboros = { workspace = true } +pipe_socket = { workspace = true } rustc-hash = { workspace = true } tempfile = { workspace = true } thiserror = { workspace = true } diff --git a/crates/fspy/src/error.rs b/crates/fspy/src/error.rs index 017d82a0e..d5162d8b3 100644 --- a/crates/fspy/src/error.rs +++ b/crates/fspy/src/error.rs @@ -16,8 +16,8 @@ pub enum SpawnError { #[error("failed to initialize seccomp_unotify supervisor: {0}")] Supervisor(std::io::Error), - #[error("failed to create IPC channel: {0}")] - ChannelCreation(std::io::Error), + #[error("failed to create IPC server: {0}")] + IpcServer(std::io::Error), /// On unix systems, the injection happens before the spawn actually occurs on. /// On Windows, the injection happens after the spawn but before resuming the process. diff --git a/crates/fspy/src/ipc.rs b/crates/fspy/src/ipc.rs index 51d498600..8669f3391 100644 --- a/crates/fspy/src/ipc.rs +++ b/crates/fspy/src/ipc.rs @@ -1,38 +1,184 @@ use std::io; -use fspy_shared::ipc::{ - PathAccess, - channel::{Receiver, ReceiverLockGuard}, +use fspy_shared::ipc::{NativeStr, PathAccess}; +use pipe_socket::{Server, ServerConnection}; +use tokio::{ + io::{AsyncReadExt as _, BufReader}, + task::{JoinHandle, JoinSet}, }; -use tokio::task::spawn_blocking; +use tokio_util::sync::CancellationToken; -// Shared memory size for storing path accesses. -// 4 GiB is large enough to store path accesses in almost any realistic scenario. -// This doesn't allocate physical memory until it's actually used. -pub const SHM_CAPACITY: usize = 4 * 1024 * 1024 * 1024; +use crate::arena::PathAccessArena; -#[ouroboros::self_referencing] -pub struct OwnedReceiverLockGuard { - /// Owns the shared memory - receiver: Receiver, - /// Borrows the shared memory and owns the file lock - #[borrows(receiver)] - #[covariant] - lock_guard: ReceiverLockGuard<'this>, +const FRAME_HEADER_LEN: usize = size_of::(); + +pub struct IpcSupervisor { + server_name: Box, + cancellation_token: CancellationToken, + task: Option>>>, +} + +impl IpcSupervisor { + pub fn bind() -> io::Result { + let server = Server::bind()?; + let server_name = server.name().into(); + let cancellation_token = CancellationToken::new(); + let task = tokio::spawn(run_server(server, cancellation_token.clone())); + Ok(Self { server_name, cancellation_token, task: Some(task) }) + } + + pub fn server_name(&self) -> &NativeStr { + &self.server_name + } + + pub async fn stop(mut self) -> io::Result> { + self.cancellation_token.cancel(); + self.task.take().expect("IPC supervisor task is missing").await.map_err(io::Error::other)? + } +} + +impl Drop for IpcSupervisor { + fn drop(&mut self) { + self.cancellation_token.cancel(); + } +} + +async fn run_server( + mut server: Server, + cancellation_token: CancellationToken, +) -> io::Result> { + let mut readers = JoinSet::new(); + let mut arenas = Vec::new(); + + loop { + tokio::select! { + biased; + () = cancellation_token.cancelled() => break, + result = readers.join_next(), if !readers.is_empty() => { + collect_reader(result.expect("reader set is not empty"), &mut arenas)?; + } + connection = server.accept() => { + readers.spawn(read_connection(connection?)); + } + } + } + + // Dropping the server prevents any new clients from connecting. Existing + // connections remain alive in their reader tasks and are drained to EOF. + drop(server); + while let Some(result) = readers.join_next().await { + collect_reader(result, &mut arenas)?; + } + Ok(arenas) } -impl OwnedReceiverLockGuard { - pub fn lock(receiver: Receiver) -> io::Result { - Self::try_new(receiver, fspy_shared::ipc::channel::Receiver::lock) +fn collect_reader( + result: Result, tokio::task::JoinError>, + arenas: &mut Vec, +) -> io::Result<()> { + arenas.push(result.map_err(io::Error::other)??); + Ok(()) +} + +async fn read_connection(connection: ServerConnection) -> io::Result { + let mut connection = BufReader::new(connection); + let mut arena = PathAccessArena::default(); + let mut frame = Vec::new(); + let mut header = [0; FRAME_HEADER_LEN]; + + loop { + if let Err(error) = connection.read_exact(&mut header).await { + if connection_closed(&error) { + return Ok(arena); + } + return Err(error); + } + + let frame_len = u32::from_le_bytes(header) as usize; + frame.resize(frame_len, 0); + if let Err(error) = connection.read_exact(&mut frame).await { + if connection_closed(&error) { + return Ok(arena); + } + return Err(error); + } + + let access: PathAccess<'_> = wincode::deserialize_exact(&frame) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + arena.add(access); + } +} + +fn connection_closed(error: &io::Error) -> bool { + matches!( + error.kind(), + io::ErrorKind::UnexpectedEof + | io::ErrorKind::BrokenPipe + | io::ErrorKind::ConnectionAborted + | io::ErrorKind::ConnectionReset + ) +} + +#[cfg(test)] +mod tests { + use std::{sync::mpsc, time::Duration}; + + use fspy_shared::ipc::{AccessMode, NativePath, PathAccessSender}; + + use super::*; + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stop_rejects_new_connections_and_waits_for_existing_ones() { + let supervisor = IpcSupervisor::bind().unwrap(); + let server_name = supervisor.server_name().to_cow_os_str().into_owned(); + let client_server_name = server_name.clone(); + let (connected_tx, connected_rx) = tokio::sync::oneshot::channel(); + let (close_tx, close_rx) = mpsc::channel(); + + let client = tokio::task::spawn_blocking(move || { + let mut sender = PathAccessSender::connect(&client_server_name).unwrap(); + sender.send(PathAccess { mode: AccessMode::READ, path: test_path() }).unwrap(); + connected_tx.send(()).unwrap(); + close_rx.recv().unwrap(); + }); + connected_rx.await.unwrap(); + + let mut stop = tokio::spawn(supervisor.stop()); + assert!(tokio::time::timeout(Duration::from_millis(50), &mut stop).await.is_err()); + let rejected = tokio::task::spawn_blocking(move || PathAccessSender::connect(&server_name)) + .await + .unwrap(); + assert!(rejected.is_err()); + + close_tx.send(()).unwrap(); + client.await.unwrap(); + let arenas = stop.await.unwrap().unwrap(); + let accesses = + arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).collect::>(); + assert_eq!(accesses.len(), 1); + assert_eq!(accesses[0].mode, AccessMode::READ); + assert_eq!(accesses[0].path, test_path()); } - pub async fn lock_async(receiver: Receiver) -> io::Result { - spawn_blocking(move || Self::lock(receiver)).await.expect("lock task panicked") + #[cfg(unix)] + fn test_path() -> &'static NativePath { + std::path::Path::new("/fspy-pipe-socket-test").into() } - pub fn iter_path_accesses(&self) -> impl Iterator> { - self.borrow_lock_guard() - .iter_frames() - .map(|frame| wincode::deserialize_exact(frame).unwrap()) + #[cfg(windows)] + fn test_path() -> &'static NativePath { + NativePath::from_wide(&[ + b'\\' as u16, + b'?' as u16, + b'?' as u16, + b'\\' as u16, + b'C' as u16, + b':' as u16, + b'\\' as u16, + b't' as u16, + b'e' as u16, + b's' as u16, + b't' as u16, + ]) } } diff --git a/crates/fspy/src/lib.rs b/crates/fspy/src/lib.rs index 6c89414ba..74a0a96e4 100644 --- a/crates/fspy/src/lib.rs +++ b/crates/fspy/src/lib.rs @@ -13,7 +13,6 @@ mod os_impl; #[path = "./windows/mod.rs"] mod os_impl; -#[cfg(unix)] mod arena; mod command; diff --git a/crates/fspy/src/unix/mod.rs b/crates/fspy/src/unix/mod.rs index f01f63b5d..d97817f81 100644 --- a/crates/fspy/src/unix/mod.rs +++ b/crates/fspy/src/unix/mod.rs @@ -8,9 +8,9 @@ use std::{io, path::Path}; #[cfg(target_os = "linux")] use fspy_seccomp_unotify::supervisor::supervise; -use fspy_shared::ipc::PathAccess; #[cfg(not(target_env = "musl"))] -use fspy_shared::ipc::{NativeStr, channel::channel}; +use fspy_shared::ipc::NativeStr; +use fspy_shared::ipc::PathAccess; #[cfg(target_os = "macos")] use fspy_shared_unix::payload::Artifacts; use fspy_shared_unix::{ @@ -25,7 +25,7 @@ use tokio::task::spawn_blocking; use tokio_util::sync::CancellationToken; #[cfg(not(target_env = "musl"))] -use crate::ipc::{OwnedReceiverLockGuard, SHM_CAPACITY}; +use crate::ipc::IpcSupervisor; use crate::{ChildTermination, Command, TrackedChild, arena::PathAccessArena, error::SpawnError}; #[derive(Debug)] @@ -78,12 +78,11 @@ impl SpyImpl { let supervisor = supervise::().map_err(SpawnError::Supervisor)?; #[cfg(not(target_env = "musl"))] - let (ipc_channel_conf, ipc_receiver) = - channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?; + let ipc_supervisor = IpcSupervisor::bind().map_err(SpawnError::IpcServer)?; let payload = Payload { #[cfg(not(target_env = "musl"))] - ipc_channel_conf, + server_name: ipc_supervisor.server_name().to_cow_os_str().into_owned().into(), #[cfg(target_os = "macos")] artifacts: self.artifacts.clone(), @@ -136,7 +135,7 @@ impl SpyImpl { stdout: child.stdout.take(), stderr: child.stderr.take(), // Keep polling for the child to exit in the background even if `wait_handle` is not awaited, - // because we need to stop the supervisor and lock the channel as soon as the child exits. + // because we need to stop accepting IPC connections as soon as the child exits. wait_handle: tokio::spawn(async move { let status = tokio::select! { status = child.wait() => status?, @@ -146,6 +145,9 @@ impl SpyImpl { } }; + #[cfg(not(target_env = "musl"))] + let mut ipc_arenas = ipc_supervisor.stop().await?; + let arenas = std::iter::once(exec_resolve_accesses); // Stop the supervisor and collect path accesses from it. #[cfg(target_os = "linux")] @@ -157,17 +159,12 @@ impl SpyImpl { .map(syscall_handler::SyscallHandler::into_arena), ); let arenas = arenas.collect::>(); - - // Lock the ipc channel after the child has exited. - // We are not interested in path accesses from descendants after the main child has exited. #[cfg(not(target_env = "musl"))] - let ipc_receiver_lock_guard = - OwnedReceiverLockGuard::lock_async(ipc_receiver).await?; - let path_accesses = PathAccessIterable { - arenas, - #[cfg(not(target_env = "musl"))] - ipc_receiver_lock_guard, + let arenas = { + ipc_arenas.extend(arenas); + ipc_arenas }; + let path_accesses = PathAccessIterable { arenas }; io::Result::Ok(ChildTermination { status, path_accesses }) }) @@ -179,23 +176,10 @@ impl SpyImpl { pub struct PathAccessIterable { arenas: Vec, - #[cfg(not(target_env = "musl"))] - ipc_receiver_lock_guard: OwnedReceiverLockGuard, } impl PathAccessIterable { pub fn iter(&self) -> impl Iterator> { - let accesses_in_arena = - self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied(); - - #[cfg(not(target_env = "musl"))] - { - let accesses_in_shm = self.ipc_receiver_lock_guard.iter_path_accesses(); - accesses_in_shm.chain(accesses_in_arena) - } - #[cfg(target_env = "musl")] - { - accesses_in_arena - } + self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied() } } diff --git a/crates/fspy/src/windows/mod.rs b/crates/fspy/src/windows/mod.rs index 8081e1298..e00014a9f 100644 --- a/crates/fspy/src/windows/mod.rs +++ b/crates/fspy/src/windows/mod.rs @@ -8,7 +8,7 @@ use std::{ use fspy_detours_sys::{DetourCopyPayloadToProcess, DetourUpdateProcessWithDll}; use fspy_shared::{ - ipc::{PathAccess, channel::channel}, + ipc::PathAccess, windows::{PAYLOAD_ID, Payload}, }; use futures_util::FutureExt; @@ -21,21 +21,18 @@ use winapi::{ use winsafe::co::{CP, WC}; use crate::{ - ChildTermination, TrackedChild, - command::Command, - error::SpawnError, - ipc::{OwnedReceiverLockGuard, SHM_CAPACITY}, + ChildTermination, TrackedChild, command::Command, error::SpawnError, ipc::IpcSupervisor, }; const INTERPOSE_CDYLIB: Artifact = artifact!("fspy_preload"); pub struct PathAccessIterable { - ipc_receiver_lock_guard: OwnedReceiverLockGuard, + arenas: Vec, } impl PathAccessIterable { pub fn iter(&self) -> impl Iterator> { - self.ipc_receiver_lock_guard.iter_path_accesses() + self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied() } } @@ -78,8 +75,9 @@ impl SpyImpl { command.creation_flags(CREATE_SUSPENDED); - let (channel_conf, receiver) = - channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?; + let ipc_supervisor = IpcSupervisor::bind().map_err(SpawnError::IpcServer)?; + let server_name: Box = + ipc_supervisor.server_name().to_cow_os_str().into_owned().into(); let mut spawn_success = false; let spawn_success = &mut spawn_success; @@ -99,7 +97,7 @@ impl SpyImpl { } let payload = Payload { - channel_conf: channel_conf.clone(), + server_name: server_name.clone(), ansi_dll_path_with_nul: ansi_dll_path_with_nul.to_bytes(), }; let payload_bytes = wincode::serialize(&payload).unwrap(); @@ -150,7 +148,7 @@ impl SpyImpl { stderr: child.stderr.take(), process_handle, // Keep polling for the child to exit in the background even if `wait_handle` is not awaited, - // because we need to stop the supervisor and lock the channel as soon as the child exits. + // because we need to stop accepting IPC connections as soon as the child exits. wait_handle: tokio::spawn(async move { let status = tokio::select! { status = child.wait() => status?, @@ -159,10 +157,8 @@ impl SpyImpl { child.wait().await? } }; - // Lock the ipc channel after the child has exited. - // We are not interested in path accesses from descendants after the main child has exited. - let ipc_receiver_lock_guard = OwnedReceiverLockGuard::lock_async(receiver).await?; - let path_accesses = PathAccessIterable { ipc_receiver_lock_guard }; + let arenas = ipc_supervisor.stop().await?; + let path_accesses = PathAccessIterable { arenas }; io::Result::Ok(ChildTermination { status, path_accesses }) }) diff --git a/crates/fspy/tests/rust_std.rs b/crates/fspy/tests/rust_std.rs index 5bdff9df0..5ff28b66b 100644 --- a/crates/fspy/tests/rust_std.rs +++ b/crates/fspy/tests/rust_std.rs @@ -21,6 +21,27 @@ async fn open_read() -> anyhow::Result<()> { Ok(()) } +#[test(tokio::test)] +async fn open_read_from_multiple_threads() -> anyhow::Result<()> { + let accesses = track_fn!((), |(): ()| { + let threads = ["thread-one", "thread-two"].map(|path| { + std::thread::spawn(move || { + let _ = File::open(path); + }) + }); + for thread in threads { + thread.join().unwrap(); + } + }) + .await?; + + let cwd = current_dir().unwrap(); + assert_contains(&accesses, cwd.join("thread-one").as_path(), AccessMode::READ); + assert_contains(&accesses, cwd.join("thread-two").as_path(), AccessMode::READ); + + Ok(()) +} + #[test(tokio::test)] async fn open_write() -> anyhow::Result<()> { let tmp_dir = tempfile::tempdir()?; diff --git a/crates/fspy_preload_unix/Cargo.toml b/crates/fspy_preload_unix/Cargo.toml index a430d1fa4..04ab577cb 100644 --- a/crates/fspy_preload_unix/Cargo.toml +++ b/crates/fspy_preload_unix/Cargo.toml @@ -9,7 +9,6 @@ crate-type = ["cdylib"] [target.'cfg(unix)'.dependencies] anyhow = { workspace = true } -wincode = { workspace = true } bstr = { workspace = true, default-features = false } ctor = { workspace = true } fspy_shared = { workspace = true } diff --git a/crates/fspy_preload_unix/src/client/mod.rs b/crates/fspy_preload_unix/src/client/mod.rs index daae12f5a..10f29ed66 100644 --- a/crates/fspy_preload_unix/src/client/mod.rs +++ b/crates/fspy_preload_unix/src/client/mod.rs @@ -2,23 +2,26 @@ pub mod convert; pub mod raw_exec; use std::{ - cell::Cell, ffi::OsStr, fmt::Debug, num::NonZeroUsize, os::unix::ffi::OsStrExt as _, - path::Path, sync::OnceLock, + cell::{Cell, RefCell}, + ffi::OsStr, + fmt::Debug, + os::unix::ffi::OsStrExt as _, + path::Path, + sync::OnceLock, }; use convert::{ToAbsolutePath, ToAccessMode}; -use fspy_shared::ipc::{PathAccess, channel::Sender}; +use fspy_shared::ipc::{PathAccess, PathAccessSender}; use fspy_shared_unix::{ exec::ExecResolveConfig, payload::EncodedPayload, spawn::{PreExec, handle_exec}, }; use raw_exec::RawExec; -use wincode::Serialize as _; pub struct Client { encoded_payload: EncodedPayload, - ipc_sender: Option, + process_id: u32, } // SAFETY: Client fields are only mutated during initialization in the ctor; after that, all access is read-only @@ -35,57 +38,71 @@ impl Debug for Client { } impl Client { - #[expect( - clippy::print_stderr, - reason = "preload library intentionally uses stderr for error reporting" - )] #[cfg(not(test))] fn from_env() -> Self { use fspy_shared_unix::payload::decode_payload_from_env; let encoded_payload = decode_payload_from_env().unwrap(); - - let ipc_sender = match encoded_payload.payload.ipc_channel_conf.sender() { - Ok(sender) => Some(sender), - Err(err) => { - // this can happen if the process is started after the root target process has exited. - // By that time the channel would have been closed in the receiver side. - // In this case we just leave a message and skip sending any path accesses. - eprintln!("fspy: failed to create ipc sender: {err}"); - None - } - }; - - Self { encoded_payload, ipc_sender } + Self { encoded_payload, process_id: std::process::id() } } - fn send(&self, mode: fspy_shared::ipc::AccessMode, path: &Path) -> anyhow::Result<()> { - let Some(ipc_sender) = &self.ipc_sender else { - // ipc channel not available, skip sending - return Ok(()); - }; + fn send(&self, mode: fspy_shared::ipc::AccessMode, path: &Path) { let path_bytes = path.as_os_str().as_bytes(); if path_bytes.starts_with(b"/dev/") || (cfg!(target_os = "linux") && (path_bytes.starts_with(b"/proc/") || path_bytes.starts_with(b"/sys/"))) { - return Ok(()); + return; } let path_access = PathAccess { mode, path: path.into() }; - let serialized_size = usize::try_from(PathAccess::serialized_size(&path_access)?) - .expect("serialized size exceeds usize"); + let process_id = std::process::id(); + + // A spawn implementation can repurpose inherited descriptors before + // calling exec. Dropping or reconnecting the inherited pipe client in + // that window could close one of the repurposed descriptors. The + // exec'd process gets fresh TLS and establishes its own client. + if PRE_EXEC_CHILD.get() + || POSIX_SPAWN_PARENT_PID + .get() + .is_some_and(|parent_process_id| parent_process_id != process_id) + { + return; + } - let frame_size = NonZeroUsize::new(serialized_size) - .expect("fspy: encoded PathAccess should never be empty"); + THREAD_CLIENT.with(|thread_client| { + // Connecting the pipe opens files and can re-enter an interposed + // function. The nested report is transport noise, so skip it. + let Ok(mut thread_client) = thread_client.try_borrow_mut() else { + return; + }; + + if matches!(&*thread_client, ThreadClient::Connected { process_id: owner, .. } if *owner != process_id) + { + // A fork inherits the calling thread's descriptor and TLS. + // Give the child process its own connection before it writes. + *thread_client = ThreadClient::Uninitialized; + } - let mut frame = ipc_sender - .claim_frame(frame_size) - .expect("fspy: failed to claim frame in shared memory"); - let mut writer: &mut [u8] = &mut frame; - PathAccess::serialize_into(&mut writer, &path_access)?; - assert_eq!(writer.len(), 0); + if matches!(&*thread_client, ThreadClient::Uninitialized) { + let server_name = self.encoded_payload.payload.server_name.to_cow_os_str(); + *thread_client = match PathAccessSender::connect(&server_name) { + Ok(sender) => ThreadClient::Connected { process_id, sender }, + Err(error) => { + report_connection_error(&error); + ThreadClient::Disconnected + } + }; + } - Ok(()) + let error = match &mut *thread_client { + ThreadClient::Connected { sender, .. } => sender.send(path_access).err(), + ThreadClient::Uninitialized | ThreadClient::Disconnected => None, + }; + if let Some(error) = error { + report_connection_error(&error); + *thread_client = ThreadClient::Disconnected; + } + }); } pub unsafe fn handle_exec( @@ -94,10 +111,16 @@ impl Client { raw_exec: RawExec, f: impl FnOnce(RawExec, Option) -> nix::Result, ) -> nix::Result { + // fork-based spawn implementations call exec after setting up the + // child's descriptors. Suppress reports in that transient child so + // the inherited TLS client is neither dropped nor reconnected before + // exec replaces it. + let _pre_exec_child = (self.process_id != std::process::id()).then(enter_pre_exec_child); + // SAFETY: raw_exec contains valid pointers to C strings and null-terminated arrays, as provided by the caller let mut exec = unsafe { raw_exec.to_exec() }; let pre_exec = handle_exec(&mut exec, config, &self.encoded_payload, |mode, path| { - self.send(mode, path).unwrap(); + self.send(mode, path); })?; RawExec::from_exec(exec, |raw_command| f(raw_command, pre_exec)) } @@ -113,9 +136,10 @@ impl Client { let () = unsafe { path.to_absolute_path(|abs_path| { let Some(abs_path) = abs_path else { - return Ok(Ok(())); + return Ok(Ok::<(), anyhow::Error>(())); }; - Ok(self.send(mode, Path::new(OsStr::from_bytes(abs_path)))) + self.send(mode, Path::new(OsStr::from_bytes(abs_path))); + Ok(Ok::<(), anyhow::Error>(())) }) }??; @@ -125,6 +149,50 @@ impl Client { static CLIENT: OnceLock = OnceLock::new(); +enum ThreadClient { + Uninitialized, + Connected { process_id: u32, sender: PathAccessSender }, + Disconnected, +} + +thread_local! { + static THREAD_CLIENT: RefCell = const { RefCell::new(ThreadClient::Uninitialized) }; + static POSIX_SPAWN_PARENT_PID: Cell> = const { Cell::new(None) }; + static PRE_EXEC_CHILD: Cell = const { Cell::new(false) }; +} + +struct PreExecChildGuard(bool); + +impl Drop for PreExecChildGuard { + fn drop(&mut self) { + PRE_EXEC_CHILD.set(self.0); + } +} + +fn enter_pre_exec_child() -> PreExecChildGuard { + PreExecChildGuard(PRE_EXEC_CHILD.replace(true)) +} + +pub struct PosixSpawnGuard(Option); + +impl Drop for PosixSpawnGuard { + fn drop(&mut self) { + POSIX_SPAWN_PARENT_PID.set(self.0); + } +} + +pub fn enter_posix_spawn() -> PosixSpawnGuard { + PosixSpawnGuard(POSIX_SPAWN_PARENT_PID.replace(Some(std::process::id()))) +} + +#[expect( + clippy::print_stderr, + reason = "preload library intentionally uses stderr for error reporting" +)] +fn report_connection_error(error: &std::io::Error) { + eprintln!("fspy: path access connection failed: {error}"); +} + // Resolving and reporting a file access can call another interposed function. // Suppress same-thread re-entry to prevent recursive access handling while // still recording accesses from other threads. diff --git a/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs b/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs index 9496b1153..52975c86d 100644 --- a/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs +++ b/crates/fspy_preload_unix/src/interceptions/spawn/posix_spawn.rs @@ -4,7 +4,7 @@ use fspy_shared_unix::exec::ExecResolveConfig; use libc::{c_char, c_int}; use crate::{ - client::{global_client, raw_exec::RawExec}, + client::{enter_posix_spawn, global_client, raw_exec::RawExec}, macros::intercept, }; @@ -49,6 +49,7 @@ unsafe fn handle_posix_spawn( RawExec { prog: file, argv: argv.cast(), envp: envp.cast() }, |raw_command, pre_exec| { let call_original = move || { + let _posix_spawn_guard = enter_posix_spawn(); original( pid, raw_command.prog, diff --git a/crates/fspy_preload_windows/src/windows/client.rs b/crates/fspy_preload_windows/src/windows/client.rs index 48933414e..401d1ea43 100644 --- a/crates/fspy_preload_windows/src/windows/client.rs +++ b/crates/fspy_preload_windows/src/windows/client.rs @@ -1,46 +1,54 @@ -use std::{cell::SyncUnsafeCell, ffi::CStr, mem::MaybeUninit}; +use std::{ + cell::{RefCell, SyncUnsafeCell}, + ffi::CStr, + mem::MaybeUninit, +}; use fspy_detours_sys::DetourCopyPayloadToProcess; use fspy_shared::{ - ipc::{PathAccess, channel::Sender}, + ipc::{PathAccess, PathAccessSender}, windows::{PAYLOAD_ID, Payload}, }; use winapi::{shared::minwindef::BOOL, um::winnt::HANDLE}; pub struct Client<'a> { payload: Payload<'a>, - ipc_sender: Option, } impl<'a> Client<'a> { pub fn from_payload_bytes(payload_bytes: &'a [u8]) -> Self { let payload: Payload<'a> = wincode::deserialize_exact(payload_bytes).unwrap(); - - let ipc_sender = match payload.channel_conf.sender() { - Ok(sender) => Some(sender), - Err(err) => { - // this can happen if the process is started after the root target process has exited. - // By that time the channel would have been closed in the receiver side. - // In this case we just leave a message and skip sending any path accesses. - #[expect( - clippy::print_stderr, - reason = "preload library uses stderr for debug diagnostics" - )] - { - eprintln!("fspy: failed to create ipc sender: {err}"); - } - None - } - }; - - Self { payload, ipc_sender } + Self { payload } } pub fn send(&self, access: PathAccess<'_>) { - let Some(sender) = &self.ipc_sender else { - return; - }; - sender.write_encoded(&access).expect("failed to send path access"); + THREAD_CLIENT.with(|thread_client| { + // Connecting the pipe triggers the same file APIs fspy detours. + // A nested report is transport noise, so skip it. + let Ok(mut thread_client) = thread_client.try_borrow_mut() else { + return; + }; + + if matches!(&*thread_client, ThreadClient::Uninitialized) { + let server_name = self.payload.server_name.to_cow_os_str(); + *thread_client = match PathAccessSender::connect(&server_name) { + Ok(sender) => ThreadClient::Connected(sender), + Err(error) => { + report_connection_error(&error); + ThreadClient::Disconnected + } + }; + } + + let error = match &mut *thread_client { + ThreadClient::Connected(sender) => sender.send(access).err(), + ThreadClient::Uninitialized | ThreadClient::Disconnected => None, + }; + if let Some(error) = error { + report_connection_error(&error); + *thread_client = ThreadClient::Disconnected; + } + }); } pub unsafe fn prepare_child_process(&self, child_handle: HANDLE) -> BOOL { @@ -62,6 +70,21 @@ impl<'a> Client<'a> { } } +enum ThreadClient { + Uninitialized, + Connected(PathAccessSender), + Disconnected, +} + +thread_local! { + static THREAD_CLIENT: RefCell = const { RefCell::new(ThreadClient::Uninitialized) }; +} + +#[expect(clippy::print_stderr, reason = "preload library uses stderr for connection diagnostics")] +fn report_connection_error(error: &std::io::Error) { + eprintln!("fspy: path access connection failed: {error}"); +} + static CLIENT: SyncUnsafeCell>> = SyncUnsafeCell::new(MaybeUninit::uninit()); diff --git a/crates/fspy_shared/Cargo.toml b/crates/fspy_shared/Cargo.toml index c854d0239..c8ba25dcd 100644 --- a/crates/fspy_shared/Cargo.toml +++ b/crates/fspy_shared/Cargo.toml @@ -9,31 +9,18 @@ publish = false wincode = { workspace = true, features = ["derive"] } bitflags = { workspace = true } bumpalo = { workspace = true } -bstr = { workspace = true } bytemuck = { workspace = true, features = ["must_cast", "derive"] } -fspy_shm = { workspace = true } native_str = { workspace = true } -thiserror = { workspace = true } -tracing = { workspace = true } -uuid = { workspace = true, features = ["v4"] } +pipe_socket = { workspace = true } vite_path = { workspace = true } [target.'cfg(target_os = "windows")'.dependencies] bytemuck = { workspace = true } winapi = { workspace = true, features = ["std"] } -[dev-dependencies] -assert2 = { workspace = true } -ctor = { workspace = true } -rustc-hash = { workspace = true } -subprocess_test = { workspace = true } -tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "time"] } - [lints] workspace = true [lib] +test = false doctest = false - -[package.metadata.cargo-shear] -ignored = ["ctor"] diff --git a/crates/fspy_shared/src/ipc/channel/mod.rs b/crates/fspy_shared/src/ipc/channel/mod.rs deleted file mode 100644 index fb1b2ef14..000000000 --- a/crates/fspy_shared/src/ipc/channel/mod.rs +++ /dev/null @@ -1,242 +0,0 @@ -//! Fast mpsc IPC channel implementation based on shared memory. - -mod shm_io; - -use std::{env::temp_dir, fs::File, io, ops::Deref, path::PathBuf}; - -use fspy_shm::{Mapping, ShmKeeper}; -pub use shm_io::FrameMut; -use shm_io::{ShmReader, ShmWriter}; -use tracing::debug; -use uuid::Uuid; -use wincode::{SchemaRead, SchemaWrite}; - -use super::NativeStr; - -/// Serializable configuration to create channel senders. -#[derive(SchemaWrite, SchemaRead, Clone, Debug)] -pub struct ChannelConf { - lock_file_path: Box, - shm_id: Box, -} - -/// Creates a mpsc IPC channel with one receiver and a `ChannelConf` that can be passed around processes and used to create multiple senders -#[expect( - clippy::missing_errors_doc, - reason = "non-vite crate: cannot use vite_str/vite_path types" -)] -pub fn channel(capacity: usize) -> io::Result<(ChannelConf, Receiver)> { - // Initialize the lock file with a unique name. - let lock_file_path = temp_dir().join(format!("fspy_ipc_{}.lock", Uuid::new_v4())); - - let (keeper, handle) = fspy_shm::create(capacity)?; - let mapping = handle.map()?; - - let conf = ChannelConf { - lock_file_path: lock_file_path.as_os_str().into(), - shm_id: keeper.id().into(), - }; - - let receiver = Receiver::new(lock_file_path, keeper, mapping)?; - Ok((conf, receiver)) -} - -impl ChannelConf { - /// Creates a sender. - /// - /// This doesn't block on the file lock. Instead it returns immediately with error if the receiver is locked or dropped. - #[expect( - clippy::missing_errors_doc, - reason = "error conditions are self-evident from return type" - )] - pub fn sender(&self) -> io::Result { - let lock_file = File::open(self.lock_file_path.to_cow_os_str())?; - lock_file.try_lock_shared()?; - - let mapping = fspy_shm::open(&self.shm_id.to_cow_os_str())?.map()?; - // SAFETY: `mapping` is a freshly mapped shared memory region with valid - // pointer and size. Exclusive write access is ensured by the shared - // file lock held by this sender. - let writer = unsafe { ShmWriter::new(mapping) }; - Ok(Sender { writer, lock_file, lock_file_path: self.lock_file_path.clone() }) - } -} - -pub struct Sender { - writer: ShmWriter, - lock_file_path: Box, - lock_file: File, -} - -impl Drop for Sender { - fn drop(&mut self) { - if let Err(err) = self.lock_file.unlock() { - debug!("Failed to unlock the shared IPC lock {:?}: {}", self.lock_file_path, err); - } - } -} - -impl Deref for Sender { - type Target = ShmWriter; - - fn deref(&self) -> &Self::Target { - &self.writer - } -} - -/// SAFETY: `Sender` holds a shared file lock that ensures there's no reader, so `shm` can be safely written to. -unsafe impl Send for Sender {} - -/// SAFETY: `Sender` holds a shared file lock that ensures there's no reader, so `shm` can be safely written to. -unsafe impl Sync for Sender {} - -/// The unique receiver side of an IPC channel. -/// Owns the lock file and removes it on drop. -pub struct Receiver { - lock_file_path: PathBuf, - lock_file: File, - /// Keeps the backing file's name alive for as long as senders may attach. - _keeper: ShmKeeper, - mapping: Mapping, -} - -/// SAFETY: `Receiver` doesn't read or write `shm`. It only passes it to `ReceiverLockGuard` under the lock. -unsafe impl Send for Receiver {} - -/// SAFETY: `Receiver` doesn't read or write `shm`. It only passes it to `ReceiverLockGuard` under the lock. -unsafe impl Sync for Receiver {} - -impl Drop for Receiver { - fn drop(&mut self) { - if let Err(err) = std::fs::remove_file(&self.lock_file_path) { - debug!("Failed to remove IPC lock file {:?}: {}", self.lock_file_path, err); - } - } -} - -impl Receiver { - fn new(lock_file_path: PathBuf, keeper: ShmKeeper, mapping: Mapping) -> io::Result { - let lock_file = File::create(&lock_file_path)?; - Ok(Self { lock_file_path, lock_file, _keeper: keeper, mapping }) - } - - /// Lock the shared memory for unique read access. - /// Blocks until all the senders have dropped (or processes owning them have all exited) so the shared memory can be safely read. - /// During the lifetime of returned `ReceiverReadGuard`, no new senders can be created (`ChannelConf::sender` would fail). - #[expect( - clippy::missing_errors_doc, - reason = "error conditions are self-evident from return type" - )] - pub fn lock(&self) -> io::Result> { - self.lock_file.lock()?; - // SAFETY: The exclusive file lock is held, so no writers can access the shared memory. - // The lock ensures all prior writes are visible to this thread. - let reader = ShmReader::new(unsafe { self.mapping.as_slice() }); - Ok(ReceiverLockGuard { reader, lock_file: &self.lock_file }) - } -} - -pub struct ReceiverLockGuard<'a> { - reader: ShmReader<&'a [u8]>, - lock_file: &'a File, -} - -impl Drop for ReceiverLockGuard<'_> { - fn drop(&mut self) { - if let Err(err) = self.lock_file.unlock() { - debug!("Failed to unlock IPC lock file: {}", err); - } - } -} -impl<'a> Deref for ReceiverLockGuard<'a> { - type Target = ShmReader<&'a [u8]>; - - fn deref(&self) -> &Self::Target { - &self.reader - } -} - -#[cfg(test)] -mod tests { - use std::{num::NonZeroUsize, str::from_utf8}; - - use bstr::B; - use subprocess_test::command_for_fn; - - use super::*; - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn smoke() { - let (conf, receiver) = channel(100).unwrap(); - let cmd = command_for_fn!(conf, |conf: ChannelConf| { - let sender = conf.sender().unwrap(); - let frame_size = NonZeroUsize::new(2).unwrap(); - let mut frame = sender.claim_frame(frame_size).unwrap(); - frame.copy_from_slice(&[4, 2]); - }); - assert!(std::process::Command::from(cmd).status().unwrap().success()); - - let lock = receiver.lock().unwrap(); - let mut frames = lock.iter_frames(); - - let received_frame = frames.next().unwrap(); - assert_eq!(received_frame, &[4, 2]); - - assert!(frames.next().is_none()); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - #[expect(clippy::print_stdout, reason = "test diagnostics")] - async fn forbid_new_senders_after_locked() { - let (conf, receiver) = channel(42).unwrap(); - let _lock = receiver.lock().unwrap(); - - let cmd = command_for_fn!(conf, |conf: ChannelConf| { - print!("{}", conf.sender().is_ok()); - }); - let output = std::process::Command::from(cmd).output().unwrap(); - assert_eq!(B(&output.stdout), B("false")); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - #[expect(clippy::print_stdout, reason = "test diagnostics")] - async fn forbid_new_senders_after_receiver_dropped() { - let (conf, receiver) = channel(42).unwrap(); - drop(receiver); - - let cmd = command_for_fn!(conf, |conf: ChannelConf| { - print!("{}", conf.sender().is_ok()); - }); - let output = std::process::Command::from(cmd).output().unwrap(); - assert_eq!(B(&output.stdout), B("false")); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn concurrent_senders() { - let (conf, receiver) = channel(8192).unwrap(); - for i in 0u16..200 { - let cmd = command_for_fn!((conf.clone(), i), |(conf, i): (ChannelConf, u16)| { - let sender = conf.sender().unwrap(); - let data_to_send = i.to_string(); - sender - .claim_frame(NonZeroUsize::new(data_to_send.len()).unwrap()) - .unwrap() - .copy_from_slice(data_to_send.as_bytes()); - }); - let output = std::process::Command::from(cmd).output().unwrap(); - assert!( - output.status.success(), - "Failed to send in iteration {}: {:?}", - i, - B(&output.stderr) - ); - } - let lock = receiver.lock().unwrap(); - let mut received_values: Vec = lock - .iter_frames() - .map(|frame| from_utf8(frame).unwrap().parse::().unwrap()) - .collect(); - received_values.sort_unstable(); - assert_eq!(received_values, (0u16..200).collect::>()); - } -} diff --git a/crates/fspy_shared/src/ipc/channel/shm_io.rs b/crates/fspy_shared/src/ipc/channel/shm_io.rs deleted file mode 100644 index 429704862..000000000 --- a/crates/fspy_shared/src/ipc/channel/shm_io.rs +++ /dev/null @@ -1,721 +0,0 @@ -//! Provides lock-free concurrent writing and reading of frames in a shared memory region. - -use core::iter::from_fn; -use std::{ - num::NonZeroUsize, - ops::{Deref, DerefMut}, - ptr::slice_from_raw_parts_mut, - sync::atomic::{AtomicI32, AtomicUsize, Ordering, fence}, -}; - -use bytemuck::must_cast; -use fspy_shm::Mapping; -use wincode::{SchemaWrite, Serialize as _, config::DefaultConfig}; - -// `ShmWriter` writes headers using atomic operations to prevent partial writes due to crashes, -// while `ShmReader` reads headers by simple pointer dereferences. -// This is safe because `ShmReader` is only used after all writing is done and visible to the calling thread (see docs of `ShmReader::new`). -// To ensure that the layouts of atomic types and their non-atomic counterparts are the same: -const _: () = { - assert!(size_of::() == size_of::()); - assert!(align_of::() == align_of::()); - assert!(size_of::() == size_of::()); - assert!(align_of::() == align_of::()); -}; - -/// A trait to borrow a raw memory region. -pub trait AsRawSlice { - fn as_raw_slice(&self) -> *mut [u8]; -} - -impl AsRawSlice for Mapping { - fn as_raw_slice(&self) -> *mut [u8] { - slice_from_raw_parts_mut(self.as_ptr(), self.len()) - } -} - -/// A concurrent shared memory writer. -/// -/// It's lock-free and safe to use across multiple threads/processes at the same time. -/// Internally it uses atomic operations to ensure that multiple writers can write to the shared memory without -/// overwriting each other's data. -pub struct ShmWriter { - /* - Layout of the whole shared memory: - | total byte size of frames(AtomicUsize) | frame 1 | frame 2 | ..... | - - Possible layout states of each frame: - - | 0(AtomicI32) | 0000...... | all zero. This happens when the thread/process crashed right after the frame is claimed. - - | byte size of the frame (AtomicI32) | partially written data | extra 0s to align to next frame header | This happens when the thread/process crashed during writing. - - | negative byte size of the frame (AtomicI32) | fully written data | extra 0s to align to next frame header | This is the normal case (negative size indicates completion). - */ - mem: M, - - #[cfg(test)] - fail_on_claim: bool, -} - -// unsafe impl Send for ShmWriter {} -// unsafe impl Sync for ShmWriter {} - -#[track_caller] -fn assert_alignment(ptr: *const u8) { - // Assert that the header of the shm is aligned to usize - assert_eq!(ptr as usize % align_of::(), 0); - // Assert that the content after whole shm header is aligned to i32 - assert_eq!((ptr as usize + size_of::()) % align_of::(), 0); -} - -const fn roundup_to_align_frame_header(mut size: usize) -> usize { - // round up new_end so that the next frame header is aligned - const FRAME_HEADER_ALIGN: usize = align_of::(); - if !size.is_multiple_of(FRAME_HEADER_ALIGN) { - size += FRAME_HEADER_ALIGN - (size % FRAME_HEADER_ALIGN); - } - size -} - -pub struct FrameMut<'a> { - header: &'a AtomicI32, - content: &'a mut [u8], -} -impl Deref for FrameMut<'_> { - type Target = [u8]; - - fn deref(&self) -> &Self::Target { - self.content - } -} -impl DerefMut for FrameMut<'_> { - fn deref_mut(&mut self) -> &mut Self::Target { - self.content - } -} - -impl Drop for FrameMut<'_> { - fn drop(&mut self) { - // Prevents compiler from ordering memory operations. Ensure the data is visible before marking as fully written - fence(Ordering::Release); - - // Mark as fully written (negative size indicates completion) - let frame_size_i32 = - i32::try_from(self.content.len()).expect("frame size checked in `append_frame`"); - self.header.store(-frame_size_i32, Ordering::Relaxed); - } -} - -#[derive(thiserror::Error, Debug)] -pub enum WriteEncodedError { - #[error("Failed to encode value into shared memory")] - EncodeError(#[from] wincode::error::WriteError), - #[error("Tried to write a frame of zero size into shared memory")] - ZeroSizedFrame, - #[error("Not enough space in shared memory to write the encoded frame")] - InsufficientSpace, -} - -impl ShmWriter { - /// Create a new `ShmWriter` backed by a shared memory region. - /// - /// # Safety - /// - `mem.as_raw_slice()` must return a stable valid pointer to a memory region of `total` bytes, - /// - the memory region must only be accessed via `ShmWriter` across all the processes. - /// - The unused region of the shared memory must be initialized to zero. - pub unsafe fn new(mem: M) -> Self { - assert_alignment(mem.as_raw_slice() as *const u8); - Self { - mem, - #[cfg(test)] - fail_on_claim: false, - } - } - - // Unwrap `self` and return the underlying memory. - #[cfg(test)] - pub fn into_memory(self) -> M { - self.mem - } - - #[cfg(test)] - const fn set_fail_on_claim(&mut self, fail_on_claim: bool) { - self.fail_on_claim = fail_on_claim; - } - - /// Claim a frame of size `frame_size`. - /// - /// Returns `None` if there is no sufficient remaining space (or simulated crash in tests) - /// `frame_size` must be non-zero because frame header being 0 would be ambiguous. - pub fn claim_frame(&self, frame_size: NonZeroUsize) -> Option> { - let shm_slice: *mut [u8] = self.mem.as_raw_slice(); - let shm_ptr = shm_slice.cast::(); - let shm_len = self.mem.as_raw_slice().len(); - - let frame_size = frame_size.get(); - let Ok(frame_size_i32) = i32::try_from(frame_size) else { - // The frame header uses a signed 32-bit integer (i32) to store the frame size. - // Negative values are reserved to indicate completion, so only positive values are valid. - // Therefore, the maximum allowed frame size is i32::MAX (2^31-1), approximately 2GB. - // Attempting to claim a frame larger than this will fail. - return None; - }; - - // Get the atomic value of the end position (first 8 bytes of shared memory) - // SAFETY: `shm_ptr` points to the start of the shared memory region, which is properly - // aligned to `usize` (verified by `assert_alignment` in `new`), and the allocation is - // large enough to contain at least a `usize` header. - let atomic_header = unsafe { AtomicUsize::from_ptr(shm_ptr.cast()) }; - - let frame_with_header_size = size_of::() + frame_size; - - // Try to atomically claim the space - // Different writers only share the header, not each other's content. so relaxed ordering is sufficient. - let current_end = - atomic_header.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current_end| { - let new_end = roundup_to_align_frame_header(current_end + frame_with_header_size); - - // Check if we have enough space - if size_of::() + new_end > shm_len { - return None; - } - - Some(new_end) - }); - - let Ok(current_end) = current_end else { - return None; // Not enough space - }; - - #[cfg(test)] - if self.fail_on_claim { - // Simulate crash right after claiming the space - return None; - } - - // Successfully claimed the space, now write the data - - // SAFETY: The atomic fetch_update above guaranteed that `size_of::() + current_end` - // is within the shared memory bounds, so this pointer arithmetic stays within the allocation. - let frame_start = unsafe { - shm_ptr.add(/* shm header */ size_of::() + current_end) - }; - - // SAFETY: `frame_start` is properly aligned to `i32` (ensured by `roundup_to_align_frame_header`) - // and points within the shared memory allocation (bounds checked by the atomic fetch_update). - let frame_header = unsafe { AtomicI32::from_ptr(frame_start.cast()) }; - - // Mark as partially written with positive size - // Atomic operations on the frame header is only for preventing partial writes of the frame header itself (possibly due to crashes), - // not for synchronization of frame contents, so relaxed ordering is sufficient - frame_header.store(frame_size_i32, Ordering::Relaxed); - - // Prevents compiler from re-ordering memory operations. Ensure the size is visible before writing the data - fence(Ordering::Release); - - // SAFETY: `frame_start` is within bounds and adding `size_of::()` skips the frame - // header to reach the content area, which is still within the claimed space. - let frame_content_ptr = unsafe { frame_start.add(size_of::()) }; // skip the frame header - Some(FrameMut { - header: frame_header, - // SAFETY: `frame_content_ptr` is valid for `frame_size` bytes (guaranteed by the - // atomic space claim), properly aligned for `u8`, and no other writer will access - // this region because each writer atomically claims a unique range. - content: unsafe { std::slice::from_raw_parts_mut(frame_content_ptr, frame_size) }, - }) - } - - /// Append an encoded value into the shared memory. - pub fn write_encoded>( - &self, - value: &T, - ) -> Result<(), WriteEncodedError> { - let serialized_size = - usize::try_from(T::serialized_size(value)?).expect("serialized size exceeds usize"); - - let Some(frame_size) = NonZeroUsize::new(serialized_size) else { - return Err(WriteEncodedError::ZeroSizedFrame); - }; - let Some(mut frame) = self.claim_frame(frame_size) else { - return Err(WriteEncodedError::InsufficientSpace); - }; - - let mut writer: &mut [u8] = &mut frame; - T::serialize_into(&mut writer, value)?; - assert_eq!(writer.len(), 0); - - Ok(()) - } - - #[cfg(test)] - pub fn try_write_frame(&self, frame: &[u8]) -> bool { - let Some(frame_size) = NonZeroUsize::new(frame.len()) else { - return false; - }; - let Some(mut frame_mut) = self.claim_frame(frame_size) else { - return false; - }; - frame_mut.copy_from_slice(frame); - true - } -} - -/// Reader of frames in shared memory created by `ShmWriter`. -pub struct ShmReader> { - mem: M, -} - -impl> ShmReader { - /// The content of `mem` should be created by `ShmWriter`. - /// Failing to do so may result in panics (mostly out-of-bounds), but won't trigger undefined behavior. - /// - /// The `ShmReader` must be created after all writing to the shared memory is done and visible to the calling thread. - /// This is guaranteed by `M: AsRef<[u8]>`, which means the memory region is immutable during the lifetime of `ShmReader`, - /// so no need to mark `ShmReader::new` as unsafe, but care must be taken to create a safe `M` from the shared memory. - pub fn new(mem: M) -> Self { - assert_alignment(mem.as_ref().as_ptr()); - Self { mem } - } - - /// Iterate over all the frames in the shared memory. - pub fn iter_frames(&self) -> impl Iterator { - let mem = self.mem.as_ref(); - let (header, content) = mem - .split_first_chunk::<{ size_of::() }>() - .expect("mem too small to contain header"); - let content_size: usize = must_cast(*header); - let mut remaining_content = &content[..content_size]; - - from_fn(move || { - let frame_size = loop { - // looking for the next valid frame - let (frame_header, next_remaining_content) = - remaining_content.split_first_chunk::<{ size_of::() }>()?; - remaining_content = next_remaining_content; - let frame_header: i32 = must_cast(*frame_header); - match frame_header { - 0 => { - // frame was claimed but never written (crashed process) - // Keep reading until we find a non-zero header - } - 1.. => { - // Partially written frame - skip it and continue - let size = usize::try_from(frame_header).unwrap(); - remaining_content = - &remaining_content[roundup_to_align_frame_header(size)..]; - } - ..0 => { - // Fully written frame (negative size indicates completion) - break usize::try_from(-frame_header).unwrap(); - } - } - }; - - let (frame_with_padding, next_remaining_content) = - remaining_content.split_at(roundup_to_align_frame_header(frame_size)); - remaining_content = next_remaining_content; - - Some(&frame_with_padding[..frame_size]) - }) - } -} - -#[cfg(test)] -mod tests { - use std::{ - process::{Child, Command}, - sync::Arc, - thread, - }; - - use assert2::assert; - use bstr::BStr; - use rustc_hash::FxHashSet; - - use super::*; - - /// A mocked shared memory region for testing. - /// - /// To be testable for miri, the shared memory is allocated using `Arc` instead of real shared memory APIs. - #[derive(Clone)] - struct MockedShm { - // Why usize: to ensure alignment - // - // Why not Arc<[usize]>: - // According to miri, from the perspective of data racing, incrementing ref count of Arc<[T]> - // is considered the same as reading the content of [T], which conflicts with writing to [T] by `ShmWriter`. - // This problem is unrelated to real shared memory. - mem: Arc>, - /// The actual requested byte length. - /// - /// over-allocation might happen to ensure alignment of `usize`, so `mem.len()` might be inaccurate. - len: usize, - } - // SAFETY: `MockedShm` uses `Arc>` for its backing memory, which is safe to send - // across threads. The raw pointer access through `AsRawSlice` is synchronized by `ShmWriter`'s - // atomic operations. - unsafe impl Send for MockedShm {} - // SAFETY: Concurrent access to the shared memory is synchronized by `ShmWriter`'s atomic - // operations. The `Arc` wrapper ensures the allocation remains valid. - unsafe impl Sync for MockedShm {} - impl MockedShm { - fn alloc(len: usize) -> Self { - // allocates this many of usize to fit the requested byte size - let size_in_usize = len / size_of::() + 1; - - let mem: Vec = std::iter::repeat_n(0usize, size_in_usize).collect(); - - Self { mem: Arc::new(mem), len } - } - } - impl AsRef<[u8]> for MockedShm { - fn as_ref(&self) -> &[u8] { - // SAFETY: `Vec::as_ptr` returns a valid pointer to the vec's buffer. The vec is - // allocated with enough `usize` elements to cover `self.len` bytes, and the pointer - // is valid for reads of `self.len` bytes. The `Arc` ensures the allocation is alive. - unsafe { std::slice::from_raw_parts(Vec::as_ptr(&self.mem).cast(), self.len) } - } - } - - impl AsRawSlice for MockedShm { - fn as_raw_slice(&self) -> *mut [u8] { - slice_from_raw_parts_mut(Vec::as_ptr(&self.mem).cast::().cast_mut(), self.len) - } - } - - #[test] - fn single_thread_basic() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_frame(b"hello")); - assert!(writer.try_write_frame(b"world")); - assert!(writer.try_write_frame(b"this is a test")); - assert!(!writer.try_write_frame(&vec![0u8; 2048])); // too large - - let reader = ShmReader::new(writer.into_memory()); - let mut frames = reader.iter_frames(); - assert_eq!(frames.next().unwrap(), b"hello"); - assert_eq!(frames.next().unwrap(), b"world"); - assert_eq!(frames.next().unwrap(), b"this is a test"); - assert_eq!(frames.next(), None); - } - #[test] - fn single_thread_empty() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_frame(b"hello")); - assert!(!writer.try_write_frame(b"")); - assert!(writer.try_write_frame(b"this is a test")); - - let reader = ShmReader::new(writer.into_memory()); - let mut frames = reader.iter_frames(); - assert_eq!(frames.next().unwrap(), b"hello"); - assert_eq!(frames.next().unwrap(), b"this is a test"); - assert_eq!(frames.next(), None); - } - - #[test] - fn single_thread_crash_after_claim() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_frame(b"foo")); - - // Simulate crash during writing - writer.set_fail_on_claim(true); - assert!(!writer.try_write_frame(b"hello")); - - writer.set_fail_on_claim(false); - assert!(writer.try_write_frame(b"bar")); - - let reader = ShmReader::new(writer.into_memory()); - let mut frames = reader.iter_frames(); - assert_eq!(frames.next().unwrap(), b"foo"); - assert_eq!(frames.next().unwrap(), b"bar"); - assert_eq!(frames.next(), None); - } - - #[test] - fn single_thread_crash_partial_write() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - assert!(writer.try_write_frame(b"foo")); - - // Simulate crash during writing - let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); - frame[..3].copy_from_slice(b"wor"); - std::mem::forget(frame); - - assert!(writer.try_write_frame(b"bar")); - - let reader = ShmReader::new(writer.into_memory()); - let mut frames = reader.iter_frames(); - assert_eq!(frames.next().unwrap(), b"foo"); - assert_eq!(frames.next().unwrap(), b"bar"); - assert_eq!(frames.next(), None); - } - - #[test] - fn single_thread_two_crashes_after_claim_and_partial_write() { - // This test verifies that ShmReader::iter correctly handles MULTIPLE consecutive - // invalid frames by continuing the loop. It's crucial for testing - // that the reader doesn't stop at the first invalid frame but keeps processing - // through multiple crash scenarios to find valid frames beyond them. - - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - - assert!(writer.try_write_frame(b"foo")); - - // First crash: AfterClaim (leaves frame header as 0) - writer.set_fail_on_claim(true); - assert!(!writer.try_write_frame(b"world")); - writer.set_fail_on_claim(false); - - // Second crash: PartialWrite (leaves positive frame header) - let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); - frame[..3].copy_from_slice(b"wor"); - std::mem::forget(frame); - - assert!(writer.try_write_frame(b"bar")); - - // ShmReader must skip BOTH invalid frames (0 header + partial header) - // and find the valid frame beyond them - this tests the loop continuation - - let reader = ShmReader::new(writer.into_memory()); - let mut frames = reader.iter_frames(); - assert_eq!(frames.next().unwrap(), b"foo"); - assert_eq!(frames.next().unwrap(), b"bar"); - assert_eq!(frames.next(), None); - } - - #[test] - fn single_thread_two_crashes_partial_write_and_after_claim() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let mut writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - // This test verifies the same loop continuation behavior but with crashes - // in reverse order. This ensures the loop correctly handles different - // sequences of invalid frame types (partial write -> after claim). - - assert!(writer.try_write_frame(b"foo")); - - // First crash: PartialWrite (leaves positive frame header) - let mut frame = writer.claim_frame(5.try_into().unwrap()).unwrap(); - frame[..3].copy_from_slice(b"wor"); - std::mem::forget(frame); - - // Second crash: AfterClaim (leaves frame header as 0) - writer.set_fail_on_claim(true); - assert!(!writer.try_write_frame(b"world")); - writer.set_fail_on_claim(false); - - assert!(writer.try_write_frame(b"bar")); - - let reader = ShmReader::new(writer.into_memory()); - // ShmReader must skip BOTH invalid frames in this order and continue - // processing to find valid frames - tests loop robustness - let mut frames = reader.iter_frames(); - assert_eq!(frames.next().unwrap(), b"foo"); - assert_eq!(frames.next().unwrap(), b"bar"); - assert_eq!(frames.next(), None); - } - - #[test] - fn concurrent() { - let shm = MockedShm::alloc(1024 * 4); - - thread::scope(|s| { - for _ in 0..4 { - s.spawn(|| { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized - // allocation. The clone shares the same backing memory, which is safe because - // `ShmWriter` uses atomic operations for concurrent access. - let writer = unsafe { ShmWriter::new(shm.clone()) }; - for _ in 0..10 { - assert!(writer.try_write_frame(b"hello")); - assert!(writer.try_write_frame(b"foo")); - assert!(writer.try_write_frame(b"this is a test")); - } - }); - } - }); - let mut count = 0; - let reader = ShmReader::new(shm); - for frame in reader.iter_frames() { - count += 1; - let frame = BStr::new(frame); - assert!(frame == b"hello" || frame == b"foo" || frame == b"this is a test"); - } - assert_eq!(count, 120); - } - - #[test] - fn concurrent_exceeded_size() { - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - thread::scope(|s| { - for _ in 0..4 { - s.spawn(|| { - for _ in 0..10 { - writer.try_write_frame(b"hello"); - writer.try_write_frame(b"foo"); - writer.try_write_frame(b"this is a test"); - } - }); - } - }); - let mut count = 0; - let reader = ShmReader::new(writer.into_memory()); - for frame in reader.iter_frames() { - count += 1; - let frame = BStr::new(frame); - assert!(frame == b"hello" || frame == b"foo" || frame == b"this is a test"); - } - assert!(count > 50); - } - - #[test] - fn test_integer_overflow_space_calculation() { - // Test case for potential integer overflow in space calculation - - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(1024)) }; - - // Try to trigger integer overflow by using maximum values - let large_frame = vec![0u8; (i32::MAX as usize) - 100]; - - // This should fail safely, not cause overflow - assert!(!writer.try_write_frame(&large_frame)); - - // Small frame should still work - assert!(writer.try_write_frame(b"test")); - - let reader = ShmReader::new(writer.into_memory()); - let mut frames = reader.iter_frames(); - assert_eq!(frames.next().unwrap(), b"test"); - assert_eq!(frames.next(), None); - } - - #[test] - fn test_space_calculation_race_condition() { - // Test for race condition in space calculation where multiple threads - // might calculate overlapping space requirements - - // SAFETY: `MockedShm::alloc` provides a valid, properly-sized, zero-initialized allocation. - let writer = unsafe { ShmWriter::new(MockedShm::alloc(200)) }; - - // Very small buffer - thread::scope(|s| { - for _ in 0..10 { - s.spawn(|| { - // Many threads trying to write large-ish frames - writer - .try_write_frame(b"this_is_a_moderately_long_frame_that_might_cause_races"); - }); - } - }); - - // The exact count doesn't matter, but the reader should not panic - // and should handle any race conditions gracefully - - let reader = ShmReader::new(writer.into_memory()); - let mut count = 0; - for _frame in reader.iter_frames() { - count += 1; - } - // At least some but not all writes should succeed - assert!(count > 0); - assert!(count < 10); - } - - #[test] - fn test_alignment_violation_detection() { - struct Misaligned(MockedShm); - impl AsRawSlice for Misaligned { - fn as_raw_slice(&self) -> *mut [u8] { - let raw_slice = self.0.as_raw_slice(); - slice_from_raw_parts_mut( - // SAFETY: Adding 1 byte to create a deliberately misaligned pointer for testing. - // The original allocation is large enough that adding 1 byte stays within bounds. - unsafe { raw_slice.cast::().add(1) }, - raw_slice.len() - 1, - ) - } - } - // Test that alignment violations are properly detected - - // Allocate memory with proper alignment first - let shm = MockedShm::alloc(64); - - // Create a deliberately misaligned pointer by adding 1 byte - // This ensures the pointer is NOT aligned to usize boundary - let misaligned_shm = Misaligned(shm); - - // Verify the pointer is actually misaligned - assert_ne!(misaligned_shm.as_raw_slice().cast::() as usize % align_of::(), 0); - - // This should panic due to alignment assertion - let result = std::panic::catch_unwind(|| { - // SAFETY: Intentionally passing a misaligned pointer to test that the alignment - // assertion in `ShmWriter::new` correctly panics. This is expected to panic. - unsafe { ShmWriter::new(misaligned_shm) }; - }); - - // Verify that the alignment check properly caught the violation - assert!(result.is_err(), "Should panic on misaligned pointer"); - } - - #[test] - #[cfg(not(miri))] - fn real_shm_across_processes() { - use subprocess_test::command_for_fn; - - const CHILD_COUNT: usize = 12; - const FRAME_COUNT_EACH_CHILD: usize = 100; - - const SHM_SIZE: usize = 1024 * 1024; - - let (keeper, handle) = fspy_shm::create(SHM_SIZE).unwrap(); - let shm_name = keeper.id().to_str().expect("test temp dir is UTF-8").to_owned(); - // Map before the children run. Windows keeps views coherent while they - // exist at the same time; a view created after every writer exited can - // observe the file before the writers' dirty pages reach it. - let mapping = handle.map().unwrap(); - - let children: Vec = (0..CHILD_COUNT) - .map(|child_index| { - let cmd = command_for_fn!( - (shm_name.clone(), child_index), - |(shm_name, child_index): (String, usize)| { - let mapping = - fspy_shm::open(std::ffi::OsStr::new(&shm_name)).unwrap().map().unwrap(); - // SAFETY: `mapping` is a freshly mapped shared memory region with a - // valid pointer and size. Concurrent write access is safe because - // `ShmWriter` uses atomic operations. - let writer = unsafe { ShmWriter::new(mapping) }; - for i in 0..FRAME_COUNT_EACH_CHILD { - let frame_data = std::format!("{child_index} {i}"); - assert!(writer.try_write_frame(frame_data.as_bytes())); - } - } - ); - Command::from(cmd).spawn().unwrap() - }) - .collect(); - - for mut c in children { - let status = c.wait().unwrap(); - assert!(status.success()); - } - - // SAFETY: All child processes have exited (waited above), so no concurrent writers exist. - // The shared memory is valid and fully written. - let shm = unsafe { mapping.as_slice() }; - let reader = ShmReader::new(shm); - let frames = reader.iter_frames().map(BStr::new).collect::>(); - assert_eq!(frames.len(), CHILD_COUNT * FRAME_COUNT_EACH_CHILD); - for child_index in 0..CHILD_COUNT { - for i in 0..FRAME_COUNT_EACH_CHILD { - let frame_data = format!("{child_index} {i}"); - assert!(frames.contains(&BStr::new(frame_data.as_bytes()))); - } - } - } -} diff --git a/crates/fspy_shared/src/ipc/mod.rs b/crates/fspy_shared/src/ipc/mod.rs index c7236e5d6..80a2dd374 100644 --- a/crates/fspy_shared/src/ipc/mod.rs +++ b/crates/fspy_shared/src/ipc/mod.rs @@ -1,11 +1,13 @@ -#[cfg(not(target_env = "musl"))] -pub mod channel; mod native_path; +#[cfg(not(target_env = "musl"))] +mod path_access_sender; use std::fmt::Debug; use bitflags::bitflags; pub use native_path::NativePath; pub use native_str::NativeStr; +#[cfg(not(target_env = "musl"))] +pub use path_access_sender::PathAccessSender; use wincode::{SchemaRead, SchemaWrite}; #[derive(SchemaWrite, SchemaRead, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)] diff --git a/crates/fspy_shared/src/ipc/path_access_sender.rs b/crates/fspy_shared/src/ipc/path_access_sender.rs new file mode 100644 index 000000000..8c47f1234 --- /dev/null +++ b/crates/fspy_shared/src/ipc/path_access_sender.rs @@ -0,0 +1,54 @@ +use std::{ffi::OsStr, io, io::Write as _}; + +use pipe_socket::Client; +use wincode::Serialize as _; + +use super::PathAccess; + +const FRAME_HEADER_LEN: usize = size_of::(); + +/// A synchronous, single-threaded sender for framed path-access records. +pub struct PathAccessSender { + client: Client, + frame: Vec, +} + +impl PathAccessSender { + /// Connects to an fspy supervisor. + /// + /// # Errors + /// + /// Returns an error if the pipe socket connection cannot be established. + pub fn connect(server_name: &OsStr) -> io::Result { + Ok(Self { client: Client::connect(server_name)?, frame: Vec::new() }) + } + + /// Serializes and sends one path-access record. + /// + /// # Errors + /// + /// Returns an error if the framed record cannot be written to the pipe. + /// + /// # Panics + /// + /// Panics if the serialized record is larger than `u32::MAX` bytes or if + /// serialization produces an inconsistent size. + pub fn send(&mut self, access: PathAccess<'_>) -> io::Result<()> { + let payload_len = usize::try_from( + PathAccess::serialized_size(&access).expect("failed to size PathAccess"), + ) + .expect("serialized PathAccess size exceeds usize"); + let payload_len_u32 = + u32::try_from(payload_len).expect("serialized PathAccess size exceeds u32"); + + self.frame.clear(); + self.frame.extend_from_slice(&payload_len_u32.to_le_bytes()); + self.frame.resize(FRAME_HEADER_LEN + payload_len, 0); + + let mut payload = &mut self.frame[FRAME_HEADER_LEN..]; + PathAccess::serialize_into(&mut payload, &access).expect("failed to serialize PathAccess"); + debug_assert!(payload.is_empty()); + + self.client.write_all(&self.frame) + } +} diff --git a/crates/fspy_shared/src/windows/mod.rs b/crates/fspy_shared/src/windows/mod.rs index cf7c536be..acd8aea19 100644 --- a/crates/fspy_shared/src/windows/mod.rs +++ b/crates/fspy_shared/src/windows/mod.rs @@ -1,7 +1,7 @@ use winapi::DEFINE_GUID; use wincode::{SchemaRead, SchemaWrite}; -use crate::ipc::channel::ChannelConf; +use crate::ipc::NativeStr; // Generated by guidgen.exe // {FC4845F1-3A8B-4F05-A3D3-A5E9E102AF33} @@ -22,6 +22,6 @@ DEFINE_GUID!( #[derive(SchemaWrite, SchemaRead, Debug, Clone)] pub struct Payload<'a> { - pub channel_conf: ChannelConf, + pub server_name: Box, pub ansi_dll_path_with_nul: &'a [u8], } diff --git a/crates/fspy_shared_unix/src/payload.rs b/crates/fspy_shared_unix/src/payload.rs index 5267bd42e..7a382eb31 100644 --- a/crates/fspy_shared_unix/src/payload.rs +++ b/crates/fspy_shared_unix/src/payload.rs @@ -4,14 +4,12 @@ use base64::{Engine as _, prelude::BASE64_STANDARD_NO_PAD}; use bstr::BString; #[cfg(not(target_env = "musl"))] use fspy_shared::ipc::NativeStr; -#[cfg(not(target_env = "musl"))] -use fspy_shared::ipc::channel::ChannelConf; use wincode::{SchemaRead, SchemaWrite}; #[derive(Debug, SchemaWrite, SchemaRead)] pub struct Payload { #[cfg(not(target_env = "musl"))] - pub ipc_channel_conf: ChannelConf, + pub server_name: Box, #[cfg(not(target_env = "musl"))] pub preload_path: Box, diff --git a/crates/pipe_socket/src/windows.rs b/crates/pipe_socket/src/windows.rs index d871a5a8e..0581dab87 100644 --- a/crates/pipe_socket/src/windows.rs +++ b/crates/pipe_socket/src/windows.rs @@ -19,6 +19,10 @@ pub struct Server { /// and created before the accepted instance is handed out, so concurrent /// connect attempts never find the pipe without an instance. pending: NamedPipeServer, + /// A separate instance that no client connects to. Its presence lets a + /// client distinguish a busy data pipe from a server that stopped + /// accepting while older connections are still draining. + _liveness: NamedPipeServer, } impl Server { @@ -29,7 +33,9 @@ impl Server { )] let name = OsString::from(format!(r"\\.\pipe\pipe_socket_{}", uuid::Uuid::new_v4())); let pending = ServerOptions::new().first_pipe_instance(true).create(&name)?; - Ok(Self { name, pending }) + let liveness = + ServerOptions::new().first_pipe_instance(true).create(liveness_name(&name))?; + Ok(Self { name, pending, _liveness: liveness }) } pub fn name(&self) -> &OsStr { @@ -83,31 +89,40 @@ impl Client { /// Opens the named pipe as a client. /// /// Opening fails with `ERROR_PIPE_BUSY` when another client claimed the - /// server's only pending instance moments earlier, in the window between - /// the server accepting one connection and creating the next instance. - /// `WaitNamedPipeW` hands that wait to the kernel: it blocks until an - /// instance is available and fails when the pipe is gone. No polling and - /// no arbitrary timeouts. + /// server's pending instance moments earlier, in the window before the + /// server creates the next instance. Short `WaitNamedPipeW` waits hand + /// that contention to the kernel while periodically checking the separate + /// liveness pipe. There is no total timeout: a live server can remain busy + /// indefinitely, while a stopped server is detected even if accepted data + /// connections still exist. pub fn connect(name: &OsStr) -> io::Result { // ERROR_PIPE_BUSY, from WinError.h. `std::io::Error` has no typed // constant for it. const ERROR_PIPE_BUSY: i32 = 231; - // NMPWAIT_WAIT_FOREVER, from WinBase.h. winapi 0.3 does not define - // the NMPWAIT_* constants. - const NMPWAIT_WAIT_FOREVER: u32 = 0xFFFF_FFFF; + // ERROR_SEM_TIMEOUT, from WinError.h. + const ERROR_SEM_TIMEOUT: i32 = 121; + const LIVENESS_CHECK_INTERVAL_MS: u32 = 100; - let mut wide: Vec = name.encode_wide().collect(); - wide.push(0); + let wide = wide_name(name); + let liveness_wide = wide_name(&liveness_name(name)); loop { match std::fs::OpenOptions::new().read(true).write(true).open(name) { Ok(inner) => return Ok(Self { inner }), Err(err) if err.raw_os_error() == Some(ERROR_PIPE_BUSY) => { + if !server_is_alive(&liveness_wide)? { + return Err(server_is_gone()); + } + // SAFETY: `wide` is NUL-terminated and remains valid for // the duration of the call. - let ok = unsafe { WaitNamedPipeW(wide.as_ptr(), NMPWAIT_WAIT_FOREVER) }; - if ok == 0 { - return Err(io::Error::last_os_error()); + let available = + unsafe { WaitNamedPipeW(wide.as_ptr(), LIVENESS_CHECK_INTERVAL_MS) }; + if available == 0 { + let error = io::Error::last_os_error(); + if error.raw_os_error() != Some(ERROR_SEM_TIMEOUT) { + return Err(error); + } } } Err(err) => return Err(err), @@ -116,6 +131,36 @@ impl Client { } } +fn liveness_name(name: &OsStr) -> OsString { + let mut name = name.to_os_string(); + name.push("_liveness"); + name +} + +fn wide_name(name: &OsStr) -> Vec { + name.encode_wide().chain([0]).collect() +} + +fn server_is_alive(liveness_wide: &[u16]) -> io::Result { + // ERROR_FILE_NOT_FOUND, from WinError.h. + const ERROR_FILE_NOT_FOUND: i32 = 2; + + // The liveness pipe always has one unclaimed instance, so a zero-timeout + // wait succeeds while the server owns it and reports file-not-found after + // the server is dropped or its process exits. + // SAFETY: `liveness_wide` is NUL-terminated and valid for this call. + if unsafe { WaitNamedPipeW(liveness_wide.as_ptr(), 0) } != 0 { + return Ok(true); + } + + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(ERROR_FILE_NOT_FOUND) { Ok(false) } else { Err(error) } +} + +fn server_is_gone() -> io::Error { + io::Error::new(io::ErrorKind::ConnectionRefused, "IPC server is gone") +} + impl Read for Client { fn read(&mut self, buf: &mut [u8]) -> io::Result { self.inner.read(buf) diff --git a/crates/pipe_socket/tests/integration.rs b/crates/pipe_socket/tests/integration.rs index a81c4532f..9622285c6 100644 --- a/crates/pipe_socket/tests/integration.rs +++ b/crates/pipe_socket/tests/integration.rs @@ -1,6 +1,7 @@ -use std::io::{Read as _, Write as _}; -#[cfg(unix)] -use std::sync::mpsc; +use std::{ + io::{Read as _, Write as _}, + sync::mpsc, +}; use pipe_socket::{Client, Server}; use tokio::{ @@ -173,3 +174,34 @@ fn connect_fails_when_server_is_gone() { } }); } + +/// Dropping the listener must reject a new client even while an accepted +/// connection remains open for draining. +#[test] +fn connect_fails_after_listener_stops_with_connection_open() { + let runtime = Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async { + let mut server = Server::bind().expect("bind server"); + let name = server.name().to_owned(); + let (close_tx, close_rx) = mpsc::channel(); + + let client = tokio::task::spawn_blocking(move || { + let _client = Client::connect(&name).expect("connect first client"); + close_rx.recv().expect("wait to close first client"); + }); + let connection = server.accept().await.expect("accept first client"); + let stale_name = server.name().to_owned(); + drop(server); + + let late_client = tokio::task::spawn_blocking(move || Client::connect(&stale_name)); + let result = tokio::time::timeout(std::time::Duration::from_secs(10), late_client) + .await + .expect("late connect must not hang") + .expect("late client task panicked"); + assert!(result.is_err()); + + close_tx.send(()).expect("close first client"); + client.await.expect("first client task panicked"); + drop(connection); + }); +}