Skip to content

Commit e1d4fa0

Browse files
wan9chiGPT-5.6
andcommitted
refactor(fspy): add shared-memory service lifecycle
Co-authored-by: GPT-5.6 <gpt-5.6@openai.com>
1 parent 3d5e984 commit e1d4fa0

6 files changed

Lines changed: 185 additions & 21 deletions

File tree

crates/fspy/src/unix/mod.rs

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,10 @@ use std::{io, path::Path};
1010
use fspy_seccomp_unotify::supervisor::supervise;
1111
use fspy_shared::ipc::PathAccess;
1212
#[cfg(not(target_env = "musl"))]
13-
use fspy_shared::ipc::{NativeStr, channel::channel};
13+
use fspy_shared::ipc::{
14+
NativeStr,
15+
channel::{CreatedChannel, channel},
16+
};
1417
#[cfg(target_os = "macos")]
1518
use fspy_shared_unix::payload::Artifacts;
1619
use fspy_shared_unix::{
@@ -78,8 +81,12 @@ impl SpyImpl {
7881
let supervisor = supervise::<SyscallHandler>().map_err(SpawnError::Supervisor)?;
7982

8083
#[cfg(not(target_env = "musl"))]
81-
let (ipc_channel_conf, ipc_receiver) =
82-
channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?;
84+
let CreatedChannel {
85+
conf: ipc_channel_conf,
86+
receiver: ipc_receiver,
87+
#[cfg(target_os = "linux")]
88+
broker: shm_broker,
89+
} = channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?;
8390

8491
let payload = Payload {
8592
#[cfg(not(target_env = "musl"))]
@@ -123,6 +130,9 @@ impl SpyImpl {
123130
});
124131
}
125132

133+
#[cfg(all(target_os = "linux", not(target_env = "musl")))]
134+
let shm_broker_handle = shm_broker.start();
135+
126136
// tokio_command.spawn blocks while executing the `pre_exec` closure.
127137
// Run it inside spawn_blocking to avoid blocking the tokio runtime, especially the supervisor loop,
128138
// which needs to accept incoming connections while `pre_exec` is connecting to it.
@@ -146,6 +156,9 @@ impl SpyImpl {
146156
}
147157
};
148158

159+
#[cfg(all(target_os = "linux", not(target_env = "musl")))]
160+
shm_broker_handle.stop().await?;
161+
149162
let arenas = std::iter::once(exec_resolve_accesses);
150163
// Stop the supervisor and collect path accesses from it.
151164
#[cfg(target_os = "linux")]

crates/fspy/src/windows/mod.rs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,10 @@ use std::{
88

99
use fspy_detours_sys::{DetourCopyPayloadToProcess, DetourUpdateProcessWithDll};
1010
use fspy_shared::{
11-
ipc::{PathAccess, channel::channel},
11+
ipc::{
12+
PathAccess,
13+
channel::{CreatedChannel, channel},
14+
},
1215
windows::{PAYLOAD_ID, Payload},
1316
};
1417
use futures_util::FutureExt;
@@ -78,7 +81,7 @@ impl SpyImpl {
7881

7982
command.creation_flags(CREATE_SUSPENDED);
8083

81-
let (channel_conf, receiver) =
84+
let CreatedChannel { conf: channel_conf, receiver } =
8285
channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?;
8386

8487
let mut spawn_success = false;
Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
#![cfg(all(target_os = "linux", not(target_env = "musl")))]
2+
3+
use std::{env::current_exe, io, process::Stdio};
4+
5+
use fspy::error::SpawnError;
6+
use fspy_shared::ipc::channel::{CreatedChannel, channel};
7+
use tokio_util::sync::CancellationToken;
8+
9+
#[test_log::test(tokio::test)]
10+
async fn broker_stops_explicitly() -> anyhow::Result<()> {
11+
let CreatedChannel { broker, .. } = channel(64 * 1024)?;
12+
broker.start().stop().await?;
13+
Ok(())
14+
}
15+
16+
#[test_log::test(tokio::test)]
17+
async fn injection_failure_drops_unstarted_broker() -> anyhow::Result<()> {
18+
let mut command = fspy::Command::new(current_exe()?);
19+
command.arg("--help").stdout(Stdio::null()).env("FSPY_PAYLOAD", "conflict");
20+
21+
let result = command.spawn(CancellationToken::new()).await;
22+
assert!(matches!(result, Err(SpawnError::Injection(_))));
23+
Ok(())
24+
}
25+
26+
#[test_log::test(tokio::test)]
27+
async fn spawn_failure_drops_started_broker() -> anyhow::Result<()> {
28+
let mut command = fspy::Command::new(current_exe()?);
29+
command.arg("--help").stdout(Stdio::null());
30+
// SAFETY: The closure only constructs an OS error and returns without touching process state.
31+
unsafe {
32+
command.pre_exec(|| Err(io::Error::from_raw_os_error(libc::EINVAL)));
33+
}
34+
35+
let result = command.spawn(CancellationToken::new()).await;
36+
assert!(matches!(result, Err(SpawnError::OsSpawn(_))));
37+
Ok(())
38+
}

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

Lines changed: 38 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@ mod shm_io;
44

55
use std::{env::temp_dir, fs::File, io, mem::MaybeUninit, ops::Deref, path::PathBuf, sync::Arc};
66

7-
use fspy_shm::Shm;
7+
#[cfg(target_os = "linux")]
8+
pub use fspy_shm::ShmBroker;
9+
use fspy_shm::{CreatedShm, Shm};
810
pub use shm_io::FrameMut;
911
use shm_io::{ShmReader, ShmWriter};
1012
use tracing::debug;
@@ -54,16 +56,31 @@ pub struct ChannelConf {
5456
shm_size: usize,
5557
}
5658

59+
/// A newly created IPC channel and its platform service.
60+
pub struct CreatedChannel {
61+
/// The serializable configuration used to create senders.
62+
pub conf: ChannelConf,
63+
/// The receiving side of the channel.
64+
pub receiver: Receiver,
65+
/// The Linux service that makes the channel mapping available to senders.
66+
#[cfg(target_os = "linux")]
67+
pub broker: ShmBroker,
68+
}
69+
5770
/// Creates a mpsc IPC channel with one receiver and a `ChannelConf` that can be passed around processes and used to create multiple senders
5871
#[expect(
5972
clippy::missing_errors_doc,
6073
reason = "non-vite crate: cannot use vite_str/vite_path types"
6174
)]
62-
pub fn channel(capacity: usize) -> io::Result<(ChannelConf, Receiver)> {
75+
pub fn channel(capacity: usize) -> io::Result<CreatedChannel> {
6376
// Initialize the lock file with a unique name.
6477
let lock_file_path = temp_dir().join(format!("fspy_ipc_{}.lock", Uuid::new_v4()));
6578

66-
let shm = fspy_shm::create(capacity)?;
79+
let CreatedShm {
80+
shm,
81+
#[cfg(target_os = "linux")]
82+
broker,
83+
} = fspy_shm::create(capacity)?;
6784

6885
let conf = ChannelConf {
6986
lock_file_path: lock_file_path.as_os_str().into(),
@@ -72,7 +89,12 @@ pub fn channel(capacity: usize) -> io::Result<(ChannelConf, Receiver)> {
7289
};
7390

7491
let receiver = Receiver::new(lock_file_path, shm)?;
75-
Ok((conf, receiver))
92+
Ok(CreatedChannel {
93+
conf,
94+
receiver,
95+
#[cfg(target_os = "linux")]
96+
broker,
97+
})
7698
}
7799

78100
impl ChannelConf {
@@ -206,7 +228,9 @@ mod tests {
206228

207229
#[test]
208230
fn smoke() {
209-
let (conf, receiver) = channel(100).unwrap();
231+
let created = channel(100).unwrap();
232+
let conf = created.conf;
233+
let receiver = created.receiver;
210234
let cmd = command_for_fn!(conf, |conf: ChannelConf| {
211235
let sender = conf.sender().unwrap();
212236
let frame_size = NonZeroUsize::new(2).unwrap();
@@ -227,7 +251,9 @@ mod tests {
227251
#[test]
228252
#[expect(clippy::print_stdout, reason = "test diagnostics")]
229253
fn forbid_new_senders_after_locked() {
230-
let (conf, receiver) = channel(42).unwrap();
254+
let created = channel(42).unwrap();
255+
let conf = created.conf;
256+
let receiver = created.receiver;
231257
let _lock = receiver.lock().unwrap();
232258

233259
let cmd = command_for_fn!(conf, |conf: ChannelConf| {
@@ -240,7 +266,9 @@ mod tests {
240266
#[test]
241267
#[expect(clippy::print_stdout, reason = "test diagnostics")]
242268
fn forbid_new_senders_after_receiver_dropped() {
243-
let (conf, receiver) = channel(42).unwrap();
269+
let created = channel(42).unwrap();
270+
let conf = created.conf;
271+
let receiver = created.receiver;
244272
drop(receiver);
245273

246274
let cmd = command_for_fn!(conf, |conf: ChannelConf| {
@@ -252,7 +280,9 @@ mod tests {
252280

253281
#[test]
254282
fn concurrent_senders() {
255-
let (conf, receiver) = channel(8192).unwrap();
283+
let created = channel(8192).unwrap();
284+
let conf = created.conf;
285+
let receiver = created.receiver;
256286
for i in 0u16..200 {
257287
let cmd = command_for_fn!((conf.clone(), i), |(conf, i): (ChannelConf, u16)| {
258288
let sender = conf.sender().unwrap();

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

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -672,7 +672,10 @@ mod tests {
672672

673673
const SHM_SIZE: usize = 1024 * 1024;
674674

675-
let shm = fspy_shm::create(SHM_SIZE).unwrap();
675+
let created = fspy_shm::create(SHM_SIZE).unwrap();
676+
#[cfg(target_os = "linux")]
677+
let broker_handle = created.broker.start();
678+
let shm = created.shm;
676679
let shm_name = shm.id().to_owned();
677680

678681
let children: Vec<Child> = (0..CHILD_COUNT)
@@ -700,6 +703,9 @@ mod tests {
700703
assert!(status.success());
701704
}
702705

706+
#[cfg(target_os = "linux")]
707+
drop(broker_handle);
708+
703709
// SAFETY: All child processes have exited (waited above), so no concurrent writers exist.
704710
// The shared memory is valid and fully written.
705711
let shm = unsafe { shm.as_slice() };

crates/fspy_shm/src/lib.rs

Lines changed: 81 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
//! Behavior-neutral shared-memory facade for fspy channels.
22
3-
use std::io;
3+
use std::{future::Future, io};
44

55
use shared_memory::{Shmem, ShmemConf};
66

@@ -9,18 +9,85 @@ pub struct Shm {
99
inner: Shmem,
1010
}
1111

12+
/// A newly created shared-memory mapping and its platform service.
13+
pub struct CreatedShm {
14+
/// The owned shared-memory mapping.
15+
pub shm: Shm,
16+
/// The service that makes this mapping available to other processes.
17+
#[cfg(target_os = "linux")]
18+
pub broker: ShmBroker,
19+
}
20+
21+
/// The unstarted service for a Linux shared-memory mapping.
22+
///
23+
/// The current `shared_memory` adapter needs no broker. This no-op type reserves
24+
/// the task-scoped lifecycle used by the native Linux implementation.
25+
#[cfg(target_os = "linux")]
26+
pub struct ShmBroker {
27+
_private: (),
28+
}
29+
30+
#[cfg(target_os = "linux")]
31+
impl ShmBroker {
32+
/// Starts serving this mapping until the returned handle is stopped or dropped.
33+
pub const fn start(self) -> ShmBrokerHandle {
34+
let Self { _private: () } = self;
35+
ShmBrokerHandle { stopped: false }
36+
}
37+
}
38+
39+
/// A running Linux shared-memory service.
40+
///
41+
/// Dropping this handle stops the service as a failure-path backstop. Call
42+
/// [`Self::stop`] to perform an orderly shutdown and observe shutdown errors.
43+
#[cfg(target_os = "linux")]
44+
#[must_use = "dropping the handle stops the shared-memory service"]
45+
pub struct ShmBrokerHandle {
46+
stopped: bool,
47+
}
48+
49+
#[cfg(target_os = "linux")]
50+
impl ShmBrokerHandle {
51+
/// Stops the service and waits for it to finish.
52+
///
53+
/// # Errors
54+
///
55+
/// Returns an error if the service cannot shut down cleanly.
56+
pub fn stop(mut self) -> impl Future<Output = io::Result<()>> {
57+
self.stop_inner();
58+
std::future::ready(Ok(()))
59+
}
60+
61+
const fn stop_inner(&mut self) {
62+
if !self.stopped {
63+
self.stopped = true;
64+
}
65+
}
66+
}
67+
68+
#[cfg(target_os = "linux")]
69+
impl Drop for ShmBrokerHandle {
70+
fn drop(&mut self) {
71+
self.stop_inner();
72+
}
73+
}
74+
1275
/// Creates a shared-memory mapping of `size` bytes.
1376
///
1477
/// # Errors
1578
///
1679
/// Returns an error if the platform cannot create or map the region.
17-
pub fn create(size: usize) -> io::Result<Shm> {
80+
pub fn create(size: usize) -> io::Result<CreatedShm> {
1881
let conf = ShmemConf::new().size(size);
1982
#[cfg(target_os = "windows")]
2083
let conf = conf.allow_raw(true);
2184

2285
let inner = conf.create().map_err(io::Error::other)?;
23-
Ok(Shm { inner })
86+
Ok(CreatedShm {
87+
shm: Shm { inner },
88+
#[cfg(target_os = "linux")]
89+
broker: ShmBroker { _private: () },
90+
})
2491
}
2592

2693
/// Opens the shared-memory mapping identified by `id`.
@@ -83,7 +150,7 @@ mod tests {
83150

84151
#[test]
85152
fn create_and_open_are_shared() {
86-
let owner = create(SIZE).unwrap();
153+
let owner = create(SIZE).unwrap().shm;
87154
assert_eq!(owner.len(), SIZE);
88155
assert_eq!(owner.as_ptr() as usize % align_of::<usize>(), 0);
89156
// SAFETY: No writes occur while this slice is borrowed.
@@ -101,7 +168,7 @@ mod tests {
101168

102169
#[test]
103170
fn mapping_is_visible_across_processes() {
104-
let owner = create(SIZE).unwrap();
171+
let owner = create(SIZE).unwrap().shm;
105172
write_byte(&owner, 0, 17);
106173

107174
let command = command_for_fn!(owner.id().to_owned(), |id: String| {
@@ -115,7 +182,7 @@ mod tests {
115182

116183
#[test]
117184
fn owner_drop_prevents_new_opens() {
118-
let owner = create(SIZE).unwrap();
185+
let owner = create(SIZE).unwrap().shm;
119186
let id = owner.id().to_owned();
120187
drop(owner);
121188

@@ -124,7 +191,7 @@ mod tests {
124191

125192
#[test]
126193
fn opened_mapping_survives_owner_drop() {
127-
let owner = create(SIZE).unwrap();
194+
let owner = create(SIZE).unwrap().shm;
128195
let id = owner.id().to_owned();
129196
let opened = open(&id, SIZE).unwrap();
130197
write_byte(&owner, 0, 17);
@@ -136,6 +203,13 @@ mod tests {
136203
assert_eq!(read_byte(&opened, SIZE - 1), 29);
137204
}
138205

206+
#[cfg(target_os = "linux")]
207+
#[test]
208+
fn broker_handle_drop_is_safe() {
209+
let created = create(SIZE).unwrap();
210+
drop(created.broker.start());
211+
}
212+
139213
fn read_byte(shm: &Shm, index: usize) -> u8 {
140214
assert!(index < shm.len());
141215
// SAFETY: The index is in bounds and tests synchronize all accesses.

0 commit comments

Comments
 (0)