refactor(fspy): add shared-memory service lifecycle

Co-authored-by: GPT-5.6 <gpt-5.6@openai.com>

Author: wan9chi

crates/fspy/src/unix/mod.rs

@@ -10,7 +10,10 @@ use std::{io, path::Path};
 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,
+    channel::{CreatedChannel, channel},
+};
 #[cfg(target_os = "macos")]
 use fspy_shared_unix::payload::Artifacts;
 use fspy_shared_unix::{
@@ -28,6 +31,40 @@ use tokio_util::sync::CancellationToken;
 use crate::ipc::{OwnedReceiverLockGuard, SHM_CAPACITY};
 use crate::{ChildTermination, Command, TrackedChild, arena::PathAccessArena, error::SpawnError};
 
+#[cfg(all(target_os = "linux", not(target_env = "musl")))]
+struct AbortOnDropHandle<T> {
+    handle: Option<tokio::task::JoinHandle<T>>,
+}
+
+#[cfg(all(target_os = "linux", not(target_env = "musl")))]
+impl<T> AbortOnDropHandle<T> {
+    const fn new(handle: tokio::task::JoinHandle<T>) -> Self {
+        Self { handle: Some(handle) }
+    }
+}
+
+#[cfg(all(target_os = "linux", not(target_env = "musl")))]
+impl AbortOnDropHandle<io::Result<()>> {
+    async fn abort_and_wait(mut self) -> io::Result<()> {
+        let handle = self.handle.take().expect("broker task handle must be present");
+        handle.abort();
+        match handle.await {
+            Ok(result) => result,
+            Err(error) if error.is_cancelled() => Ok(()),
+            Err(error) => Err(io::Error::other(error)),
+        }
+    }
+}
+
+#[cfg(all(target_os = "linux", not(target_env = "musl")))]
+impl<T> Drop for AbortOnDropHandle<T> {
+    fn drop(&mut self) {
+        if let Some(handle) = &self.handle {
+            handle.abort();
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct SpyImpl {
     #[cfg(target_os = "macos")]
@@ -78,8 +115,15 @@ impl SpyImpl {
         let supervisor = supervise::<SyscallHandler>().map_err(SpawnError::Supervisor)?;
 
         #[cfg(not(target_env = "musl"))]
-        let (ipc_channel_conf, ipc_receiver) =
-            channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?;
+        let CreatedChannel {
+            conf: ipc_channel_conf,
+            receiver: ipc_receiver,
+            #[cfg(target_os = "linux")]
+                broker: shm_broker,
+        } = channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?;
+
+        #[cfg(all(target_os = "linux", not(target_env = "musl")))]
+        let shm_broker_handle = AbortOnDropHandle::new(tokio::spawn(shm_broker));
 
         let payload = Payload {
             #[cfg(not(target_env = "musl"))]
@@ -146,6 +190,9 @@ impl SpyImpl {
                     }
                 };
 
+                #[cfg(all(target_os = "linux", not(target_env = "musl")))]
+                shm_broker_handle.abort_and_wait().await?;
+
                 let arenas = std::iter::once(exec_resolve_accesses);
                 // Stop the supervisor and collect path accesses from it.
                 #[cfg(target_os = "linux")]

crates/fspy/src/windows/mod.rs

@@ -8,7 +8,10 @@ use std::{
 
 use fspy_detours_sys::{DetourCopyPayloadToProcess, DetourUpdateProcessWithDll};
 use fspy_shared::{
-    ipc::{PathAccess, channel::channel},
+    ipc::{
+        PathAccess,
+        channel::{CreatedChannel, channel},
+    },
     windows::{PAYLOAD_ID, Payload},
 };
 use futures_util::FutureExt;
@@ -78,7 +81,7 @@ impl SpyImpl {
 
         command.creation_flags(CREATE_SUSPENDED);
 
-        let (channel_conf, receiver) =
+        let CreatedChannel { conf: channel_conf, receiver } =
             channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?;
 
         let mut spawn_success = false;

crates/fspy/tests/service_lifecycle.rs

@@ -0,0 +1,42 @@
+#![cfg(all(target_os = "linux", not(target_env = "musl")))]
+
+use std::{env::current_exe, io, process::Stdio};
+
+use fspy::error::SpawnError;
+use fspy_shared::ipc::channel::{CreatedChannel, channel};
+use tokio_util::sync::CancellationToken;
+
+#[test_log::test(tokio::test)]
+async fn broker_runs_until_aborted() -> anyhow::Result<()> {
+    let CreatedChannel { broker, .. } = channel(64 * 1024)?;
+    let handle = tokio::spawn(broker);
+    tokio::task::yield_now().await;
+    assert!(!handle.is_finished());
+    handle.abort();
+    assert!(handle.await.unwrap_err().is_cancelled());
+    Ok(())
+}
+
+#[test_log::test(tokio::test)]
+async fn injection_failure_aborts_broker() -> anyhow::Result<()> {
+    let mut command = fspy::Command::new(current_exe()?);
+    command.arg("--help").stdout(Stdio::null()).env("FSPY_PAYLOAD", "conflict");
+
+    let result = command.spawn(CancellationToken::new()).await;
+    assert!(matches!(result, Err(SpawnError::Injection(_))));
+    Ok(())
+}
+
+#[test_log::test(tokio::test)]
+async fn spawn_failure_aborts_broker() -> anyhow::Result<()> {
+    let mut command = fspy::Command::new(current_exe()?);
+    command.arg("--help").stdout(Stdio::null());
+    // SAFETY: The closure only constructs an OS error and returns without touching process state.
+    unsafe {
+        command.pre_exec(|| Err(io::Error::from_raw_os_error(libc::EINVAL)));
+    }
+
+    let result = command.spawn(CancellationToken::new()).await;
+    assert!(matches!(result, Err(SpawnError::OsSpawn(_))));
+    Ok(())
+}

crates/fspy_shared/src/ipc/channel/mod.rs

@@ -4,7 +4,9 @@ mod shm_io;
 
 use std::{env::temp_dir, fs::File, io, mem::MaybeUninit, ops::Deref, path::PathBuf, sync::Arc};
 
-use fspy_shm::Shm;
+#[cfg(target_os = "linux")]
+pub use fspy_shm::ShmBroker;
+use fspy_shm::{CreatedShm, Shm};
 pub use shm_io::FrameMut;
 use shm_io::{ShmReader, ShmWriter};
 use tracing::debug;
@@ -54,16 +56,31 @@ pub struct ChannelConf {
     shm_size: usize,
 }
 
+/// A newly created IPC channel and its platform service.
+pub struct CreatedChannel {
+    /// The serializable configuration used to create senders.
+    pub conf: ChannelConf,
+    /// The receiving side of the channel.
+    pub receiver: Receiver,
+    /// The Linux service future that makes the channel mapping available to senders.
+    #[cfg(target_os = "linux")]
+    pub broker: ShmBroker,
+}
+
 /// 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)> {
+pub fn channel(capacity: usize) -> io::Result<CreatedChannel> {
     // Initialize the lock file with a unique name.
     let lock_file_path = temp_dir().join(format!("fspy_ipc_{}.lock", Uuid::new_v4()));
 
-    let shm = fspy_shm::create(capacity)?;
+    let CreatedShm {
+        shm,
+        #[cfg(target_os = "linux")]
+        broker,
+    } = fspy_shm::create(capacity)?;
 
     let conf = ChannelConf {
         lock_file_path: lock_file_path.as_os_str().into(),
@@ -72,7 +89,12 @@ pub fn channel(capacity: usize) -> io::Result<(ChannelConf, Receiver)> {
     };
 
     let receiver = Receiver::new(lock_file_path, shm)?;
-    Ok((conf, receiver))
+    Ok(CreatedChannel {
+        conf,
+        receiver,
+        #[cfg(target_os = "linux")]
+        broker,
+    })
 }
 
 impl ChannelConf {
@@ -206,7 +228,9 @@ mod tests {
 
     #[test]
     fn smoke() {
-        let (conf, receiver) = channel(100).unwrap();
+        let created = channel(100).unwrap();
+        let conf = created.conf;
+        let receiver = created.receiver;
         let cmd = command_for_fn!(conf, |conf: ChannelConf| {
             let sender = conf.sender().unwrap();
             let frame_size = NonZeroUsize::new(2).unwrap();
@@ -227,7 +251,9 @@ mod tests {
     #[test]
     #[expect(clippy::print_stdout, reason = "test diagnostics")]
     fn forbid_new_senders_after_locked() {
-        let (conf, receiver) = channel(42).unwrap();
+        let created = channel(42).unwrap();
+        let conf = created.conf;
+        let receiver = created.receiver;
         let _lock = receiver.lock().unwrap();
 
         let cmd = command_for_fn!(conf, |conf: ChannelConf| {
@@ -240,7 +266,9 @@ mod tests {
     #[test]
     #[expect(clippy::print_stdout, reason = "test diagnostics")]
     fn forbid_new_senders_after_receiver_dropped() {
-        let (conf, receiver) = channel(42).unwrap();
+        let created = channel(42).unwrap();
+        let conf = created.conf;
+        let receiver = created.receiver;
         drop(receiver);
 
         let cmd = command_for_fn!(conf, |conf: ChannelConf| {
@@ -252,7 +280,9 @@ mod tests {
 
     #[test]
     fn concurrent_senders() {
-        let (conf, receiver) = channel(8192).unwrap();
+        let created = channel(8192).unwrap();
+        let conf = created.conf;
+        let receiver = created.receiver;
         for i in 0u16..200 {
             let cmd = command_for_fn!((conf.clone(), i), |(conf, i): (ChannelConf, u16)| {
                 let sender = conf.sender().unwrap();

crates/fspy_shared/src/ipc/channel/shm_io.rs

@@ -672,7 +672,10 @@ mod tests {
 
         const SHM_SIZE: usize = 1024 * 1024;
 
-        let shm = fspy_shm::create(SHM_SIZE).unwrap();
+        let created = fspy_shm::create(SHM_SIZE).unwrap();
+        #[cfg(target_os = "linux")]
+        let _broker = created.broker;
+        let shm = created.shm;
         let shm_name = shm.id().to_owned();
 
         let children: Vec<Child> = (0..CHILD_COUNT)

crates/fspy_shm/src/lib.rs

@@ -1,6 +1,8 @@
 //! Behavior-neutral shared-memory facade for fspy channels.
 
 use std::io;
+#[cfg(target_os = "linux")]
+use std::{future::Future, pin::Pin};
 
 use shared_memory::{Shmem, ShmemConf};
 
@@ -9,18 +11,35 @@ pub struct Shm {
     inner: Shmem,
 }
 
+/// A Linux service future that makes a shared-memory mapping available to other processes.
+#[cfg(target_os = "linux")]
+pub type ShmBroker = Pin<Box<dyn Future<Output = io::Result<()>> + Send + 'static>>;
+
+/// A newly created shared-memory mapping and its platform service.
+pub struct CreatedShm {
+    /// The owned shared-memory mapping.
+    pub shm: Shm,
+    /// The service future that makes this mapping available to other processes.
+    #[cfg(target_os = "linux")]
+    pub broker: ShmBroker,
+}
+
 /// Creates a shared-memory mapping of `size` bytes.
 ///
 /// # Errors
 ///
 /// Returns an error if the platform cannot create or map the region.
-pub fn create(size: usize) -> io::Result<Shm> {
+pub fn create(size: usize) -> io::Result<CreatedShm> {
     let conf = ShmemConf::new().size(size);
     #[cfg(target_os = "windows")]
     let conf = conf.allow_raw(true);
 
     let inner = conf.create().map_err(io::Error::other)?;
-    Ok(Shm { inner })
+    Ok(CreatedShm {
+        shm: Shm { inner },
+        #[cfg(target_os = "linux")]
+        broker: Box::pin(std::future::pending()),
+    })
 }
 
 /// Opens the shared-memory mapping identified by `id`.
@@ -83,7 +102,7 @@ mod tests {
 
     #[test]
     fn create_and_open_are_shared() {
-        let owner = create(SIZE).unwrap();
+        let owner = create(SIZE).unwrap().shm;
         assert_eq!(owner.len(), SIZE);
         assert_eq!(owner.as_ptr() as usize % align_of::<usize>(), 0);
         // SAFETY: No writes occur while this slice is borrowed.
@@ -101,7 +120,7 @@ mod tests {
 
     #[test]
     fn mapping_is_visible_across_processes() {
-        let owner = create(SIZE).unwrap();
+        let owner = create(SIZE).unwrap().shm;
         write_byte(&owner, 0, 17);
 
         let command = command_for_fn!(owner.id().to_owned(), |id: String| {
@@ -115,7 +134,7 @@ mod tests {
 
     #[test]
     fn owner_drop_prevents_new_opens() {
-        let owner = create(SIZE).unwrap();
+        let owner = create(SIZE).unwrap().shm;
         let id = owner.id().to_owned();
         drop(owner);
 
@@ -124,7 +143,7 @@ mod tests {
 
     #[test]
     fn opened_mapping_survives_owner_drop() {
-        let owner = create(SIZE).unwrap();
+        let owner = create(SIZE).unwrap().shm;
         let id = owner.id().to_owned();
         let opened = open(&id, SIZE).unwrap();
         write_byte(&owner, 0, 17);
@@ -138,6 +157,14 @@ mod tests {
         assert_eq!(read_byte(&opened, SIZE - 1), 29);
     }
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn broker_remains_pending() {
+        let mut broker = create(SIZE).unwrap().broker;
+        let mut context = std::task::Context::from_waker(std::task::Waker::noop());
+        assert!(broker.as_mut().poll(&mut context).is_pending());
+    }
+
     fn read_byte(shm: &Shm, index: usize) -> u8 {
         assert!(index < shm.len());
         // SAFETY: The index is in bounds and tests synchronize all accesses.