refactor(fspy): simplify shm_io to a plain frame arena

The gated API reads shared memory only when the gate is closed and nothing
was in flight, which proves every claimed frame was fully written and
published. The three-state frame header (`0` / `+size` / `-size`), the
atomic header stores and the release fences existed so a reader could
tolerate crashed or in-flight writers — states that are unreachable through
the gate.

Delete them. A frame is now `| size: u32 | content | padding |`, written
once at claim time with a plain store, and the arena keeps exactly one
atomic: its end offset. Crash exclusion lives entirely in the gate, where a
leaked count fails closed — the run is not cached — instead of being parsed
around. The arena also stops being a surface of its own and becomes an
internal of the gated module.

No public API change: `GatedShmWriter`, `GatedShmReceiver`,
`GatedShmReader`, `FrameMut`, `ClaimError` and `StillWriting` keep their
signatures, so the channel layer, the fspy crate and both preload clients
are untouched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: wan9chi

Cargo.lock

@@ -21,7 +21,6 @@ bytemuck = { workspace = true }
 winapi = { workspace = true, features = ["std"] }
 
 [dev-dependencies]
-assert2 = { workspace = true }
 ctor = { workspace = true }
 rustc-hash = { workspace = true }
 subprocess_test = { workspace = true }

crates/fspy_shared/Cargo.toml

@@ -27,12 +27,14 @@ use std::{
     sync::{Arc, atomic::AtomicU64},
 };
 
-use wincode::{SchemaWrite, config::DefaultConfig};
+use wincode::{SchemaWrite, Serialize as _, config::DefaultConfig};
 
-use super::{
-    gate::{EnterError, Gate, GateGuard},
-    shm_io::{self, AsRawSlice, ShmReader, ShmWriter},
-};
+use self::shm_io::{ShmReader, ShmWriter};
+use super::gate::{EnterError, Gate, GateGuard};
+
+mod shm_io;
+
+pub use shm_io::AsRawSlice;
 
 /// Bytes reserved in front of the arena for the gate word.
 const GATE_REGION_SIZE: usize = 64;
@@ -125,29 +127,45 @@ impl<M: AsRawSlice> GatedShmWriter<M> {
     /// [`ClaimError::Capacity`] when the arena cannot fit the frame.
     pub fn claim_frame(&self, size: NonZeroUsize) -> Result<FrameMut<'_>, ClaimError> {
         let gate = self.gate().enter()?;
-        let frame = self.arena.claim_frame(size).ok_or(ClaimError::Capacity)?;
-        Ok(FrameMut { frame, _gate: gate })
+        let content = self.arena.claim_frame(size).ok_or(ClaimError::Capacity)?;
+        Ok(FrameMut { content, _gate: gate })
     }
 
     /// Appends an encoded value as a single frame.
     ///
     /// # Errors
     ///
     /// Returns [`ClaimError::Closed`] wrapped in [`WriteEncodedError::Claim`]
-    /// once the read end has closed the gate, and otherwise whatever the arena
-    /// reports.
+    /// once the read end has closed the gate, [`WriteEncodedError::ZeroSizedFrame`]
+    /// for a value that encodes to nothing, and otherwise the encoder's error.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the encoder does not fill exactly the size it asked for, which
+    /// would mean the schema is inconsistent with itself.
     pub fn write_encoded<T: SchemaWrite<DefaultConfig, Src = T>>(
         &self,
         value: &T,
     ) -> Result<(), WriteEncodedError> {
-        // The guard is held across the inner write and released only afterwards,
-        // so the frame is complete before the gate count drops.
-        let _gate = self.gate().enter().map_err(ClaimError::from)?;
-        self.arena.write_encoded(value).map_err(|error| match error {
-            shm_io::WriteEncodedError::EncodeError(error) => WriteEncodedError::Encode(error),
-            shm_io::WriteEncodedError::ZeroSizedFrame => WriteEncodedError::ZeroSizedFrame,
-            shm_io::WriteEncodedError::InsufficientSpace => ClaimError::Capacity.into(),
-        })
+        // Enter the gate before touching the value, so a post-close call
+        // reports `Claim(Closed)` no matter what the value encodes to.
+        let gate = self.gate().enter().map_err(ClaimError::from)?;
+
+        let serialized_size =
+            usize::try_from(T::serialized_size(value)?).expect("serialized size exceeds usize");
+        let Some(size) = NonZeroUsize::new(serialized_size) else {
+            return Err(WriteEncodedError::ZeroSizedFrame);
+        };
+
+        let content = self.arena.claim_frame(size).ok_or(ClaimError::Capacity)?;
+        // The frame holds the gate guard, so it is released only after the
+        // content is written.
+        let mut frame = FrameMut { content, _gate: gate };
+        let mut writer: &mut [u8] = &mut frame;
+        T::serialize_into(&mut writer, value)?;
+        assert_eq!(writer.len(), 0);
+
+        Ok(())
     }
 
     fn gate(&self) -> Gate<'_> {
@@ -158,24 +176,25 @@ impl<M: AsRawSlice> GatedShmWriter<M> {
 /// An exclusively owned frame together with the gate guard that keeps the region
 /// open while it is written.
 ///
-/// The frame is declared first so that its completion lands before the gate is
-/// released.
+/// Dropping it releases the gate, which is what publishes the content to a
+/// reader; the arena's own header was already written when the frame was
+/// claimed.
 pub struct FrameMut<'a> {
-    frame: shm_io::FrameMut<'a>,
+    content: &'a mut [u8],
     _gate: GateGuard<'a>,
 }
 
 impl Deref for FrameMut<'_> {
     type Target = [u8];
 
     fn deref(&self) -> &Self::Target {
-        &self.frame
+        self.content
     }
 }
 
 impl DerefMut for FrameMut<'_> {
     fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.frame
+        self.content
     }
 }
 
@@ -407,6 +426,18 @@ mod tests {
         assert!(reader.iter_frames().all(|frame| frame == b"filler"));
     }
 
+    #[test]
+    fn write_encoded_after_close_reports_closed_for_any_value() {
+        let (writer, receiver) = endpoints(1024);
+        let _frames = receiver.close().unwrap();
+
+        // A zero-sized value must still report the close, not its own size.
+        assert!(matches!(
+            writer.write_encoded(&()),
+            Err(WriteEncodedError::Claim(ClaimError::Closed))
+        ));
+    }
+
     #[test]
     fn write_encoded_roundtrips_through_the_gate() {
         let (writer, receiver) = endpoints(1024);
@@ -452,4 +483,162 @@ mod tests {
         assert_eq!(seen.len(), WRITERS * PER_WRITER);
         assert_eq!(seen[0], "0 0");
     }
+
+    /// Writers that run out of room must fail cleanly rather than corrupt the
+    /// arena for the ones that fitted.
+    #[test]
+    fn concurrent_writers_exceeding_the_arena() {
+        let (writer, receiver) = endpoints(1024);
+
+        thread::scope(|scope| {
+            for _ in 0..4 {
+                scope.spawn(|| {
+                    for _ in 0..10 {
+                        let _ = write_frame(&writer, b"hello");
+                        let _ = write_frame(&writer, b"foo");
+                        let _ = write_frame(&writer, b"this is a test");
+                    }
+                });
+            }
+        });
+
+        let reader = receiver.close().unwrap();
+        let mut count = 0;
+        for frame in reader.iter_frames() {
+            count += 1;
+            assert!(
+                frame == b"hello" || frame == b"foo" || frame == b"this is a test",
+                "unexpected frame {:?}",
+                from_utf8(frame)
+            );
+        }
+        assert!(count > 20, "expected most writes to fit, got {count}");
+    }
+
+    /// Several threads racing for the last few bytes must all agree on who got
+    /// them, and the reader must walk whatever survived without panicking.
+    #[test]
+    fn concurrent_writers_racing_for_the_last_bytes() {
+        let (writer, receiver) = endpoints(GATE_REGION_SIZE + 200);
+
+        thread::scope(|scope| {
+            for _ in 0..10 {
+                scope.spawn(|| {
+                    let _ = write_frame(
+                        &writer,
+                        b"this_is_a_moderately_long_frame_that_might_cause_races",
+                    );
+                });
+            }
+        });
+
+        let reader = receiver.close().unwrap();
+        let count = reader.iter_frames().count();
+        // Some but not all of them fit into 200 bytes of arena.
+        assert!(count > 0);
+        assert!(count < 10);
+    }
+
+    /// A frame larger than the size header can describe is refused, and the
+    /// arena stays usable afterwards.
+    #[test]
+    fn a_frame_too_large_for_the_header_is_refused() {
+        let (writer, receiver) = endpoints(1024);
+
+        let oversized = NonZeroUsize::new(usize::try_from(u32::MAX).unwrap() + 1).unwrap();
+        assert!(matches!(writer.claim_frame(oversized), Err(ClaimError::Capacity)));
+
+        write_frame(&writer, b"test").unwrap();
+        let reader = receiver.close().unwrap();
+        assert_eq!(reader.iter_frames().collect::<Vec<_>>(), vec![b"test".as_slice()]);
+    }
+
+    #[test]
+    fn a_misaligned_region_is_rejected() {
+        struct Misaligned(MockRegion);
+
+        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: shifting by one byte deliberately misaligns the
+                    // region and stays inside the over-allocated backing buffer.
+                    unsafe { raw_slice.cast::<u8>().add(1) },
+                    raw_slice.len() - 1,
+                )
+            }
+        }
+
+        let misaligned = Misaligned(MockRegion::alloc(1024));
+        assert_ne!(misaligned.as_raw_slice().cast::<u8>() as usize % align_of::<u64>(), 0);
+
+        let result = std::panic::catch_unwind(|| {
+            // SAFETY: the region is deliberately misaligned to check that the
+            // constructor catches it; the assertion fires before any access.
+            unsafe { GatedShmWriter::new(misaligned) };
+        });
+
+        assert!(result.is_err(), "a misaligned region must be rejected");
+    }
+
+    /// The same protocol over a real shared mapping, written by several
+    /// processes at once.
+    #[test]
+    #[cfg(not(miri))]
+    fn real_shm_across_processes() {
+        use std::process::{Child, Command};
+
+        use rustc_hash::FxHashSet;
+        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_id = keeper.id().to_str().expect("test temp dir is UTF-8").to_owned();
+        let mapping = handle.map().unwrap();
+        // SAFETY: the backing file was just created zero-initialized, the
+        // mapping stays valid while the receiver holds it, and only this
+        // protocol reaches it.
+        let receiver = unsafe { GatedShmReceiver::new(mapping) };
+
+        let children: Vec<Child> = (0..CHILD_COUNT)
+            .map(|child_index| {
+                let cmd = command_for_fn!((shm_id.clone(), child_index), |(
+                    shm_id,
+                    child_index,
+                ): (
+                    String,
+                    usize
+                )| {
+                    let mapping =
+                        fspy_shm::open(std::ffi::OsStr::new(&shm_id)).unwrap().map().unwrap();
+                    // SAFETY: the mapping was created under the same
+                    // contract and only this protocol reaches it.
+                    let writer = unsafe { GatedShmWriter::new(mapping) };
+                    for i in 0..FRAME_COUNT_EACH_CHILD {
+                        let frame_data = format!("{child_index} {i}");
+                        let size = NonZeroUsize::new(frame_data.len()).unwrap();
+                        writer.claim_frame(size).unwrap().copy_from_slice(frame_data.as_bytes());
+                    }
+                });
+                Command::from(cmd).spawn().unwrap()
+            })
+            .collect();
+
+        for mut child in children {
+            assert!(child.wait().unwrap().success());
+        }
+
+        let reader = receiver.close().unwrap();
+        let frames: FxHashSet<&str> =
+            reader.iter_frames().map(|frame| from_utf8(frame).unwrap()).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 {
+                assert!(frames.contains(format!("{child_index} {i}").as_str()));
+            }
+        }
+    }
 }