refactor(fspy): use shared memory IPC on Windows (#239)

### TL;DR

Replaced Windows pipe-based IPC with shared memory for file system access tracking, unifying the approach across platforms.

### What changed?

- Replaced Windows named pipe IPC with shared memory for tracking file system accesses
- Created a new `ipc.rs` module with common code for both Windows and Unix platforms
- Extracted `OwnedReceiverLockGuard` from Unix implementation to be reused on Windows
- Removed dependency on `dashmap` which was used for the previous Windows implementation
- Added `write_encoded` method to `ShmWriter` to directly encode values into shared memory
- Updated Windows payload structure to use `ChannelConf` instead of pipe handles

### Why make this change?

The previous implementation used different IPC mechanisms on Windows (named pipes) and Unix (shared memory). This change unifies the approach across platforms, making the code more maintainable and consistent. Shared memory is more efficient for passing large amounts of data between processes, which is important for tracking file system accesses in complex applications. The unified approach also simplifies the codebase by removing platform-specific code paths and dependencies.

Author: branchseer

Cargo.lock

@@ -0,0 +1,42 @@
+use std::io;
+
+use bincode::borrow_decode_from_slice;
+use fspy_shared::ipc::{
+    BINCODE_CONFIG, PathAccess,
+    channel::{Receiver, ReceiverLockGuard},
+};
+use tokio::task::spawn_blocking;
+
+// 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;
+
+#[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>,
+}
+
+impl OwnedReceiverLockGuard {
+    pub fn lock(receiver: Receiver) -> io::Result<Self> {
+        OwnedReceiverLockGuard::try_new(receiver, |receiver| receiver.lock())
+    }
+
+    pub async fn lock_async(receiver: Receiver) -> io::Result<Self> {
+        spawn_blocking(move || Self::lock(receiver)).await.expect("lock task panicked")
+    }
+
+    pub fn iter_path_accesses(&self) -> impl Iterator<Item = PathAccess<'_>> {
+        self.borrow_lock_guard().iter_frames().map(|frame| {
+            let (path_access, decoded_size) =
+                borrow_decode_from_slice::<PathAccess<'_>, _>(frame, BINCODE_CONFIG).unwrap();
+            assert_eq!(decoded_size, frame.len());
+            path_access
+        })
+    }
+}

crates/fspy/src/ipc.rs

@@ -4,6 +4,8 @@
 // Persist the injected DLL/shared library somewhere in the filesystem.
 mod fixture;
 
+mod ipc;
+
 #[cfg(unix)]
 #[path = "./unix/mod.rs"]
 mod os_impl;
@@ -12,6 +14,7 @@ mod os_impl;
 #[path = "./windows/mod.rs"]
 mod os_impl;
 
+#[cfg(unix)]
 mod arena;
 mod command;
 

crates/fspy/src/lib.rs

@@ -1,7 +1,3 @@
-// use std::{os::unix::process::CommandExt};
-
-// use tokio::process::Command;
-
 #[cfg(target_os = "linux")]
 mod syscall_handler;
 
@@ -10,13 +6,9 @@ mod macos_fixtures;
 
 use std::{io, path::Path};
 
-use bincode::borrow_decode_from_slice;
 #[cfg(target_os = "linux")]
 use fspy_seccomp_unotify::supervisor::supervise;
-use fspy_shared::ipc::{
-    BINCODE_CONFIG, NativeString, PathAccess,
-    channel::{Receiver, ReceiverLockGuard, channel},
-};
+use fspy_shared::ipc::{NativeString, PathAccess, channel::channel};
 #[cfg(target_os = "macos")]
 use fspy_shared_unix::payload::Fixtures;
 use fspy_shared_unix::{
@@ -27,9 +19,12 @@ use fspy_shared_unix::{
 use futures_util::FutureExt;
 #[cfg(target_os = "linux")]
 use syscall_handler::SyscallHandler;
-use tokio::task::spawn_blocking;
 
-use crate::{Command, TrackedChild, arena::PathAccessArena};
+use crate::{
+    Command, TrackedChild,
+    arena::PathAccessArena,
+    ipc::{OwnedReceiverLockGuard, SHM_CAPACITY},
+};
 
 #[derive(Debug, Clone)]
 pub struct SpyInner {
@@ -71,16 +66,6 @@ impl SpyInner {
     }
 }
 
-#[ouroboros::self_referencing]
-struct OwnedReceiverLockGuard {
-    /// Owns the shared memory
-    receiver: Receiver,
-    /// Borrows the shared memory and owns the file lock
-    #[borrows(receiver)]
-    #[covariant]
-    lock_guard: ReceiverLockGuard<'this>,
-}
-
 pub struct PathAccessIterable {
     arenas: Vec<PathAccessArena>,
     ipc_receiver_lock_guard: OwnedReceiverLockGuard,
@@ -91,22 +76,11 @@ impl PathAccessIterable {
         let accesses_in_arena =
             self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied();
 
-        let accesses_in_shm =
-            self.ipc_receiver_lock_guard.borrow_lock_guard().iter_frames().map(|frame| {
-                let (path_access, decoded_size) =
-                    borrow_decode_from_slice::<PathAccess<'_>, _>(frame, BINCODE_CONFIG).unwrap();
-                assert_eq!(decoded_size, frame.len());
-                path_access
-            });
+        let accesses_in_shm = self.ipc_receiver_lock_guard.iter_path_accesses();
         accesses_in_shm.chain(accesses_in_arena)
     }
 }
 
-// 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.
-const SHM_CAPACITY: usize = 4 * 1024 * 1024 * 1024;
-
 pub(crate) async fn spawn_impl(mut command: Command) -> io::Result<TrackedChild> {
     #[cfg(target_os = "linux")]
     let supervisor = supervise::<SyscallHandler>()?;
@@ -170,10 +144,7 @@ pub(crate) async fn spawn_impl(mut command: Command) -> io::Result<TrackedChild>
     let accesses_future = async move {
         let arenas = arenas_future.await?;
         // `receiver.lock()` blocks. Run it inside `spawn_blocking` to avoid blocking the tokio runtime.
-        let ipc_receiver_lock_guard = spawn_blocking(move || {
-            OwnedReceiverLockGuard::try_new(ipc_receiver, |receiver| receiver.lock())
-        })
-        .await??;
+        let ipc_receiver_lock_guard = OwnedReceiverLockGuard::lock_async(ipc_receiver).await?;
         Ok(PathAccessIterable { arenas, ipc_receiver_lock_guard })
     }
     .boxed();

crates/fspy/src/unix/mod.rs

@@ -1,40 +1,31 @@
 use std::{
-    ffi::{CStr, c_char, c_void},
-    fs::OpenOptions,
+    ffi::{CStr, c_char},
     io,
     os::windows::{ffi::OsStrExt, io::AsRawHandle, process::ChildExt as _},
     path::Path,
-    ptr::null_mut,
     sync::Arc,
 };
 
-use bincode::borrow_decode_from_slice;
 use const_format::formatcp;
 use fspy_detours_sys::{DetourCopyPayloadToProcess, DetourUpdateProcessWithDll};
 use fspy_shared::{
-    ipc::{BINCODE_CONFIG, PathAccess},
+    ipc::{BINCODE_CONFIG, PathAccess, channel::channel},
     windows::{PAYLOAD_ID, Payload},
 };
 use futures_util::FutureExt;
-use tokio::{
-    io::AsyncReadExt,
-    net::windows::named_pipe::{PipeMode, ServerOptions},
-};
-// use detours_sys2::{DetourAttach,};
 use winapi::{
-    shared::minwindef::{FALSE, TRUE},
-    um::{
-        handleapi::DuplicateHandle,
-        processthreadsapi::{GetCurrentProcess, ResumeThread},
-        winbase::CREATE_SUSPENDED,
-        winnt::DUPLICATE_SAME_ACCESS,
-    },
+    shared::minwindef::TRUE,
+    um::{processthreadsapi::ResumeThread, winbase::CREATE_SUSPENDED},
 };
-// use windows_sys::Win32::System::Threading::{CREATE_SUSPENDED, ResumeThread};
 use winsafe::co::{CP, WC};
 use xxhash_rust::const_xxh3::xxh3_128;
 
-use crate::{TrackedChild, arena::PathAccessArena, command::Command, fixture::Fixture};
+use crate::{
+    TrackedChild,
+    command::Command,
+    fixture::Fixture,
+    ipc::{OwnedReceiverLockGuard, SHM_CAPACITY},
+};
 
 const PRELOAD_CDYLIB_BINARY: &[u8] = include_bytes!(env!("CARGO_CDYLIB_FILE_FSPY_PRELOAD_WINDOWS"));
 const INTERPOSE_CDYLIB: Fixture = Fixture::new(
@@ -43,38 +34,14 @@ const INTERPOSE_CDYLIB: Fixture = Fixture::new(
     formatcp!("{:x}", xxh3_128(PRELOAD_CDYLIB_BINARY)),
 );
 
-fn luid() -> io::Result<u64> {
-    let mut luid = unsafe { std::mem::zeroed::<winapi::um::winnt::LUID>() };
-    let ret = unsafe { winapi::um::securitybaseapi::AllocateLocallyUniqueId(&mut luid) };
-    if ret == 0 {
-        return Err(io::Error::last_os_error());
-    }
-    Ok((u64::from(luid.HighPart as u32)) << 32 | u64::from(luid.LowPart))
-}
-
 pub struct PathAccessIterable {
-    arena: PathAccessArena,
-    // pipe_receiver: NamedPipeServer,
+    ipc_receiver_lock_guard: OwnedReceiverLockGuard,
 }
 
-const MESSAGE_MAX_LEN: usize = 4096;
-
 impl PathAccessIterable {
     pub fn iter(&self) -> impl Iterator<Item = PathAccess<'_>> {
-        self.arena.borrow_accesses().iter().copied()
+        self.ipc_receiver_lock_guard.iter_path_accesses()
     }
-    //     pub async fn next<'a>(&mut self, buf: &'a mut Vec<u8>) -> io::Result<Option<PathAccess<'a>>> {
-    //         buf.resize(MESSAGE_MAX_LEN, 0);
-    //         let n = self.pipe_receiver.read(buf.as_mut_slice()).await?;
-    //         if n == 0 {
-    //             return Ok(None);
-    //         }
-    //         let msg = &buf[..n];
-    //         let (path_access, decoded_len) =
-    //             borrow_decode_from_slice::<'_, PathAccess, _>(msg, BINCODE_CONFIG).unwrap();
-    //         assert_eq!(decoded_len, msg.len());
-    //         Ok(Some(path_access))
-    //     }
 }
 
 // pub struct TracedProcess {
@@ -110,46 +77,13 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result<TrackedChild> {
 
     command.creation_flags(CREATE_SUSPENDED);
 
-    let pipe_name = format!(r"\\.\pipe\fspy_ipc_{:x}", luid()?);
-
-    let mut pipe_receiver = ServerOptions::new()
-        .pipe_mode(PipeMode::Message)
-        .access_outbound(false)
-        .access_inbound(true)
-        .in_buffer_size(1024)
-        // .out_buffer_size(100 * 1024 * 1024)
-        .create(&pipe_name)?;
-
-    let connect_fut = pipe_receiver.connect();
-
-    let pipe_sender = OpenOptions::new().write(true).open(&pipe_name).unwrap();
-
-    connect_fut.await?;
-
-    // Temporary workaround before switching to shared memory IPC on Windows.
-    // The shared memory IPC requires `accesses_future` to be polled after the child process has exited.
-    // The test code has updated but the Windows implementation here has not yet adopted that model.
-    // As a result, accesses_future didn't get polled making the pipe_sender to block indefinitely.
-    // To unblock the pipe_sender, we spawn a separate task to keep reading pipe_receiver.
-    // This workaround can be removed once the shared memory IPC is used on Windows.
-    let accesses_future = tokio::task::spawn(async move {
-        let mut arena = PathAccessArena::default();
-
-        let mut buf = [0u8; MESSAGE_MAX_LEN];
-        loop {
-            let n = pipe_receiver.read(&mut buf).await?;
-            if n == 0 {
-                break;
-            }
-            let msg = &buf[..n];
-            let (path_access, decoded_len) =
-                borrow_decode_from_slice::<'_, PathAccess, _>(msg, BINCODE_CONFIG).unwrap();
-            assert_eq!(decoded_len, msg.len());
-            arena.add(path_access);
-        }
-        io::Result::Ok(PathAccessIterable { arena })
-    });
-    let accesses_future = async move { accesses_future.await.unwrap() }.boxed();
+    let (channel_conf, receiver) = channel(SHM_CAPACITY)?;
+
+    let accesses_future = async move {
+        let ipc_receiver_lock_guard = OwnedReceiverLockGuard::lock_async(receiver).await?;
+        io::Result::Ok(PathAccessIterable { ipc_receiver_lock_guard })
+    }
+    .boxed();
 
     // let path_access_stream = PathAccessIterable { pipe_receiver };
 
@@ -163,24 +97,8 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result<TrackedChild> {
             return Err(io::Error::last_os_error());
         }
 
-        let mut handle_in_child: *mut c_void = null_mut();
-        let ret = unsafe {
-            DuplicateHandle(
-                GetCurrentProcess(),
-                pipe_sender.as_raw_handle(),
-                process_handle,
-                &mut handle_in_child,
-                0,
-                FALSE,
-                DUPLICATE_SAME_ACCESS,
-            )
-        };
-        if ret == 0 {
-            return Err(io::Error::last_os_error());
-        }
-
         let payload = Payload {
-            pipe_handle: handle_in_child.addr(),
+            channel_conf: channel_conf.clone(),
             asni_dll_path_with_nul: asni_dll_path_with_nul.to_bytes(),
         };
         let payload_bytes = bincode::encode_to_vec(payload, BINCODE_CONFIG).unwrap();
@@ -207,6 +125,5 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result<TrackedChild> {
         Ok(std_child)
     })?;
 
-    drop(pipe_sender);
     Ok(TrackedChild { tokio_child: child, accesses_future })
 }

crates/fspy/src/windows/mod.rs

@@ -26,6 +26,3 @@ winsafe = { workspace = true }
 
 [target.'cfg(target_os = "windows")'.dev-dependencies]
 tempfile = { workspace = true }
-
-[dependencies]
-dashmap = { workspace = true }