From c61987a29d156bbad75f7cb45b643e493e8675e4 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Tue, 28 Jul 2026 17:58:21 +0800 Subject: [PATCH] fix(fspy): replace the IPC file lock with an in-mapping close gate The old quiescence protocol attached "may write" to the shared mapping but "is still writing" to a file-lock descriptor. A descendant that closes descriptors it does not recognize released the lock while keeping full write access to the mapping, so the receiver could read frames while a straggler was mutating them. Put the gate in the shared memory itself, where a writer cannot drop it while still being able to write: one atomic word admits and counts claims, and the runner's close is a single `fetch_or` at root-process exit that fences all future claims and reports whether any write was in flight. Zero in flight proves every admitted claim ran to completion and the memory is frozen; anything else means the run is conservatively not cached. Tracking now stops when the root process exits instead of waiting for lingering descendants, so a task that leaks a daemon no longer blocks the read step, and post-exit accesses are treated as what they are: racy with respect to the task's contract. Closes #544. Closes #396. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 1 + Cargo.lock | 3 - crates/fspy/src/ipc.rs | 34 -- crates/fspy/src/unix/mod.rs | 44 +- crates/fspy/src/windows/mod.rs | 38 +- crates/fspy/tests/test_utils/mod.rs | 4 +- crates/fspy_preload_unix/src/client/mod.rs | 66 ++- .../src/interceptions/exit.rs | 25 + .../src/interceptions/mod.rs | 1 + .../src/windows/client.rs | 13 +- .../src/supervisor/mod.rs | 112 ++++- crates/fspy_shared/Cargo.toml | 3 - crates/fspy_shared/src/ipc/channel/gate.rs | 232 +++++++++ crates/fspy_shared/src/ipc/channel/gated.rs | 455 ++++++++++++++++++ crates/fspy_shared/src/ipc/channel/mod.rs | 364 ++++++++------ crates/fspy_shm/README.md | 4 +- crates/vite_task/src/session/event.rs | 4 + .../src/session/execute/cache_update.rs | 74 ++- .../vite_task/src/session/reporter/summary.rs | 25 + 19 files changed, 1247 insertions(+), 255 deletions(-) create mode 100644 crates/fspy_preload_unix/src/interceptions/exit.rs create mode 100644 crates/fspy_shared/src/ipc/channel/gate.rs create mode 100644 crates/fspy_shared/src/ipc/channel/gated.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index a58ad9e8c..c002a64f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # Changelog +- **Changed** Automatic file-access tracking stops when a task's root process exits. Accesses from leftover descendants are no longer recorded, collecting the trace no longer waits for them to exit, and a run with a traced process still mid-write at that instant is not cached ([#544](https://github.com/voidzero-dev/vite-task/issues/544), [#396](https://github.com/voidzero-dev/vite-task/issues/396), [#577](https://github.com/voidzero-dev/vite-task/pull/577)). - **Fixed** `vp run` no longer fails while setting up task communication in default Codex CLI and Claude Code sandboxes that block Unix domain sockets ([#562](https://github.com/voidzero-dev/vite-task/issues/562), [#569](https://github.com/voidzero-dev/vite-task/pull/569)). - **Fixed** Automatic file-access tracking now works inside coding-agent sandboxes, including the default Codex CLI and Claude Code sandboxes ([#563](https://github.com/voidzero-dev/vite-task/issues/563), [#576](https://github.com/voidzero-dev/vite-task/pull/576)). - **Added** Tasks now run with `VP_RUN=1` set, so tools can tell they are running under `vp run` instead of being invoked directly ([#570](https://github.com/voidzero-dev/vite-task/pull/570)). diff --git a/Cargo.lock b/Cargo.lock index b6ef0549b..366d5cebb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1351,9 +1351,6 @@ dependencies = [ "rustc-hash", "subprocess_test", "thiserror 2.0.18", - "tokio", - "tracing", - "uuid", "vite_path", "winapi", "wincode", diff --git a/crates/fspy/src/ipc.rs b/crates/fspy/src/ipc.rs index 51d498600..b36b48469 100644 --- a/crates/fspy/src/ipc.rs +++ b/crates/fspy/src/ipc.rs @@ -1,38 +1,4 @@ -use std::io; - -use fspy_shared::ipc::{ - 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::try_new(receiver, fspy_shared::ipc::channel::Receiver::lock) - } - - pub async fn lock_async(receiver: Receiver) -> io::Result { - spawn_blocking(move || Self::lock(receiver)).await.expect("lock task panicked") - } - - pub fn iter_path_accesses(&self) -> impl Iterator> { - self.borrow_lock_guard() - .iter_frames() - .map(|frame| wincode::deserialize_exact(frame).unwrap()) - } -} diff --git a/crates/fspy/src/unix/mod.rs b/crates/fspy/src/unix/mod.rs index f01f63b5d..957091325 100644 --- a/crates/fspy/src/unix/mod.rs +++ b/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::{ChannelFrames, channel}, +}; #[cfg(target_os = "macos")] use fspy_shared_unix::payload::Artifacts; use fspy_shared_unix::{ @@ -25,7 +28,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::SHM_CAPACITY; use crate::{ChildTermination, Command, TrackedChild, arena::PathAccessArena, error::SpawnError}; #[derive(Debug)] @@ -40,7 +43,7 @@ pub struct SpyImpl { impl SpyImpl { /// Initialize the fs access spy by writing the preload library on disk. /// - /// On musl targets, we don't build a preload library — + /// On musl targets, we don't build a preload library; /// only seccomp-based tracking is used. pub fn init_in(#[cfg_attr(target_env = "musl", allow(unused))] dir: &Path) -> io::Result { #[cfg(not(target_env = "musl"))] @@ -158,15 +161,24 @@ impl SpyImpl { ); 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. + // Close the ipc channel now that the child has exited. We are not + // interested in path accesses from descendants after the main child + // has exited, and we do not wait for them either: closing is a + // single atomic operation that also fences out later writers. + // A close error means a traced process was still mid-write; its + // frames cannot be read safely, so the run is incomplete. #[cfg(not(target_env = "musl"))] - let ipc_receiver_lock_guard = - OwnedReceiverLockGuard::lock_async(ipc_receiver).await?; + let (shm_frames, incomplete) = ipc_receiver + .close() + .map_or_else(|_| (None, true), |frames| (Some(frames), false)); let path_accesses = PathAccessIterable { arenas, #[cfg(not(target_env = "musl"))] - ipc_receiver_lock_guard, + shm_frames, + #[cfg(not(target_env = "musl"))] + incomplete, + #[cfg(target_env = "musl")] + incomplete: false, }; io::Result::Ok(ChildTermination { status, path_accesses }) @@ -179,8 +191,12 @@ impl SpyImpl { pub struct PathAccessIterable { arenas: Vec, + /// `None` when the channel could not be frozen for reading. #[cfg(not(target_env = "musl"))] - ipc_receiver_lock_guard: OwnedReceiverLockGuard, + shm_frames: Option, + /// A traced process was still writing when the channel closed, so the + /// recorded accesses are not a complete picture of the run. + incomplete: bool, } impl PathAccessIterable { @@ -190,7 +206,8 @@ impl PathAccessIterable { #[cfg(not(target_env = "musl"))] { - let accesses_in_shm = self.ipc_receiver_lock_guard.iter_path_accesses(); + let accesses_in_shm = + self.shm_frames.iter().flat_map(ChannelFrames::iter_path_accesses); accesses_in_shm.chain(accesses_in_arena) } #[cfg(target_env = "musl")] @@ -198,4 +215,11 @@ impl PathAccessIterable { accesses_in_arena } } + + /// Whether tracking was cut short, which makes the accesses above an + /// incomplete record of the run. + #[must_use] + pub const fn is_incomplete(&self) -> bool { + self.incomplete + } } diff --git a/crates/fspy/src/windows/mod.rs b/crates/fspy/src/windows/mod.rs index 8081e1298..d902ebd53 100644 --- a/crates/fspy/src/windows/mod.rs +++ b/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::{ChannelFrames, channel}, + }, windows::{PAYLOAD_ID, Payload}, }; use futures_util::FutureExt; @@ -21,21 +24,29 @@ 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::SHM_CAPACITY, }; const INTERPOSE_CDYLIB: Artifact = artifact!("fspy_preload"); pub struct PathAccessIterable { - ipc_receiver_lock_guard: OwnedReceiverLockGuard, + /// `None` when the channel could not be frozen for reading. + shm_frames: Option, + /// A traced process was still writing when the channel closed, so the + /// recorded accesses are not a complete picture of the run. + incomplete: bool, } impl PathAccessIterable { pub fn iter(&self) -> impl Iterator> { - self.ipc_receiver_lock_guard.iter_path_accesses() + self.shm_frames.iter().flat_map(ChannelFrames::iter_path_accesses) + } + + /// Whether tracking was cut short, which makes the accesses above an + /// incomplete record of the run. + #[must_use] + pub const fn is_incomplete(&self) -> bool { + self.incomplete } } @@ -159,10 +170,15 @@ 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 }; + // Close the ipc channel now that the child has exited. We are not + // interested in path accesses from descendants after the main child + // has exited, and we do not wait for them either: closing is a + // single atomic operation that also fences out later writers. + // A close error means a traced process was still mid-write; its + // frames cannot be read safely, so the run is incomplete. + let (shm_frames, incomplete) = + receiver.close().map_or_else(|_| (None, true), |frames| (Some(frames), false)); + let path_accesses = PathAccessIterable { shm_frames, incomplete }; io::Result::Ok(ChildTermination { status, path_accesses }) }) diff --git a/crates/fspy/tests/test_utils/mod.rs b/crates/fspy/tests/test_utils/mod.rs index cfa46c4a9..96ee037b7 100644 --- a/crates/fspy/tests/test_utils/mod.rs +++ b/crates/fspy/tests/test_utils/mod.rs @@ -49,9 +49,11 @@ pub fn assert_contains( assert_eq!( expected_mode, actual_mode, - "Expected to find access to path {} with mode {:?}, but it was not found in: {:?}", + "Expected to find access to path {} with mode {:?}, but it was not found in \ + (tracking incomplete: {}): {:?}", expected_path.display(), expected_mode, + accesses.is_incomplete(), accesses.iter().collect::>() ); } diff --git a/crates/fspy_preload_unix/src/client/mod.rs b/crates/fspy_preload_unix/src/client/mod.rs index daae12f5a..5a01ea9cd 100644 --- a/crates/fspy_preload_unix/src/client/mod.rs +++ b/crates/fspy_preload_unix/src/client/mod.rs @@ -2,12 +2,23 @@ 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, + ffi::OsStr, + fmt::Debug, + num::NonZeroUsize, + os::unix::ffi::OsStrExt as _, + path::Path, + sync::{ + OnceLock, + atomic::{AtomicUsize, Ordering}, + }, }; use convert::{ToAbsolutePath, ToAccessMode}; -use fspy_shared::ipc::{PathAccess, channel::Sender}; +use fspy_shared::ipc::{ + PathAccess, + channel::{ClaimError, Sender}, +}; use fspy_shared_unix::{ exec::ExecResolveConfig, payload::EncodedPayload, @@ -78,9 +89,16 @@ impl Client { let frame_size = NonZeroUsize::new(serialized_size) .expect("fspy: encoded PathAccess should never be empty"); - let mut frame = ipc_sender - .claim_frame(frame_size) - .expect("fspy: failed to claim frame in shared memory"); + let _in_flight = SendInFlight::begin(); + let mut frame = match ipc_sender.claim_frame(frame_size) { + Ok(frame) => frame, + // The channel was closed because the traced root process exited. + // Accesses from whatever is left behind are dropped by design. + Err(ClaimError::Closed) => return Ok(()), + Err(ClaimError::Capacity) => { + panic!("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); @@ -125,6 +143,42 @@ impl Client { static CLIENT: OnceLock = OnceLock::new(); +/// Sends that are between claiming a frame and finishing it, on any thread. +static IN_FLIGHT_SENDS: AtomicUsize = AtomicUsize::new(0); + +struct SendInFlight; + +impl SendInFlight { + fn begin() -> Self { + IN_FLIGHT_SENDS.fetch_add(1, Ordering::Relaxed); + Self + } +} + +impl Drop for SendInFlight { + fn drop(&mut self) { + IN_FLIGHT_SENDS.fetch_sub(1, Ordering::Release); + } +} + +/// Waits until no send is mid-frame on any thread of this process. +/// +/// The exit interception calls this so a voluntary exit never abandons a +/// claimed frame, which would make the whole run's tracking incomplete. A send +/// lasts microseconds and never blocks, so this returns almost at once. The +/// deadline covers a thread that died mid-send without running its drop +/// (asynchronous `pthread_cancel` and the like): exit is delayed by at most +/// the deadline, and the runner treats the leaked count as an incomplete run. +pub fn drain_in_flight_sends() { + let deadline = std::time::Instant::now() + std::time::Duration::from_millis(100); + while IN_FLIGHT_SENDS.load(Ordering::Acquire) != 0 { + if std::time::Instant::now() > deadline { + return; + } + std::thread::yield_now(); + } +} + // 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/exit.rs b/crates/fspy_preload_unix/src/interceptions/exit.rs new file mode 100644 index 000000000..e8f981b65 --- /dev/null +++ b/crates/fspy_preload_unix/src/interceptions/exit.rs @@ -0,0 +1,25 @@ +//! Drains in-flight sends before a voluntary exit. +//! +//! A process may call `exit` while another of its threads is between claiming +//! a frame and finishing it. Dying there abandons the frame's gate guard, and +//! the runner then treats the whole run's tracking as incomplete. The drain +//! lasts microseconds; signals and crashes still skip it, and the runner +//! handles those by not caching the run. + +use libc::c_int; + +use crate::{client::drain_in_flight_sends, macros::intercept}; + +intercept!(exit: unsafe extern "C" fn(status: c_int) -> !); +unsafe extern "C" fn exit(status: c_int) -> ! { + drain_in_flight_sends(); + // SAFETY: forwarding to the real libc exit with the caller's status + unsafe { exit::original()(status) } +} + +intercept!(_exit: unsafe extern "C" fn(status: c_int) -> !); +unsafe extern "C" fn _exit(status: c_int) -> ! { + drain_in_flight_sends(); + // SAFETY: forwarding to the real libc _exit with the caller's status + unsafe { _exit::original()(status) } +} diff --git a/crates/fspy_preload_unix/src/interceptions/mod.rs b/crates/fspy_preload_unix/src/interceptions/mod.rs index 0d3742ea7..67ea574df 100644 --- a/crates/fspy_preload_unix/src/interceptions/mod.rs +++ b/crates/fspy_preload_unix/src/interceptions/mod.rs @@ -1,5 +1,6 @@ mod access; mod dirent; +mod exit; mod open; mod spawn; mod stat; diff --git a/crates/fspy_preload_windows/src/windows/client.rs b/crates/fspy_preload_windows/src/windows/client.rs index 48933414e..049e3c257 100644 --- a/crates/fspy_preload_windows/src/windows/client.rs +++ b/crates/fspy_preload_windows/src/windows/client.rs @@ -2,7 +2,10 @@ use std::{cell::SyncUnsafeCell, ffi::CStr, mem::MaybeUninit}; use fspy_detours_sys::DetourCopyPayloadToProcess; use fspy_shared::{ - ipc::{PathAccess, channel::Sender}, + ipc::{ + PathAccess, + channel::{ClaimError, Sender, WriteEncodedError}, + }, windows::{PAYLOAD_ID, Payload}, }; use winapi::{shared::minwindef::BOOL, um::winnt::HANDLE}; @@ -40,7 +43,13 @@ impl<'a> Client<'a> { let Some(sender) = &self.ipc_sender else { return; }; - sender.write_encoded(&access).expect("failed to send path access"); + match sender.write_encoded(&access) { + Ok(()) + // The channel was closed because the traced root process exited. + // Accesses from whatever is left behind are dropped by design. + | Err(WriteEncodedError::Claim(ClaimError::Closed)) => {} + Err(err) => panic!("failed to send path access: {err:?}"), + } } pub unsafe fn prepare_child_process(&self, child_handle: HANDLE) -> BOOL { diff --git a/crates/fspy_seccomp_unotify/src/supervisor/mod.rs b/crates/fspy_seccomp_unotify/src/supervisor/mod.rs index b1aa0eb62..511c6a726 100644 --- a/crates/fspy_seccomp_unotify/src/supervisor/mod.rs +++ b/crates/fspy_seccomp_unotify/src/supervisor/mod.rs @@ -2,8 +2,7 @@ pub mod handler; mod listener; use std::{ - convert::Infallible, - io::{self}, + io, os::{ fd::{FromRawFd, OwnedFd}, unix::ffi::OsStrExt, @@ -20,8 +19,8 @@ use passfd::tokio::FdPassingExt; use seccompiler::{BpfProgram, SeccompAction, SeccompFilter}; use tokio::{ net::UnixListener, - sync::oneshot, - task::{JoinHandle, JoinSet}, + sync::{mpsc, watch}, + task::JoinHandle, }; use tracing::{Level, span}; @@ -32,7 +31,7 @@ use crate::{ pub struct Supervisor { payload: SeccompPayload, - cancel_tx: oneshot::Sender, + cancel_tx: watch::Sender<()>, handling_loop_task: JoinHandle>>, } @@ -42,7 +41,12 @@ impl Supervisor { &self.payload } - /// Stops the supervisor and returns all handler instances. + /// Stops the supervisor and returns what every handler recorded so far. + /// + /// Does not wait for target processes to exit. A target that is still + /// alive keeps its notifications answered by a detached task, so its + /// filtered syscalls keep working; whatever it does from now on is not + /// recorded. /// /// # Panics /// Panics if the handling loop task has panicked. @@ -84,18 +88,24 @@ pub fn supervise() -> io::Re filter: bpf_filter, }; - // The oneshot channel is used to cancel the accept loop. - // The sender doesn't need to actually send anything. Drop is enough. - let (cancel_tx, mut cancel_rx) = oneshot::channel::(); + // Dropping the sender cancels the accept loop and tells every handler task + // to hand over its handler. + let (cancel_tx, cancel_rx) = watch::channel(()); let handling_loop = async move { - let mut join_set: JoinSet> = JoinSet::new(); + // Every spawned task sends exactly one message: its handler when its + // targets exited or when the supervisor stopped, or its I/O error. + let (handler_tx, mut handler_rx) = mpsc::unbounded_channel::>(); + let mut spawned = 0usize; + let mut accept_cancel = cancel_rx.clone(); loop { let accept_future = notify_listener.as_file().accept(); pin_mut!(accept_future); - let (incoming_stream, _) = match select(&mut cancel_rx, accept_future).await { - Either::Left((Err(_), _)) => break, + let cancelled = accept_cancel.changed(); + pin_mut!(cancelled); + let (incoming_stream, _) = match select(cancelled, accept_future).await { + Either::Left(_) => break, Either::Right((incoming, _)) => incoming?, }; let notify_fd = incoming_stream.recv_fd().await?; @@ -106,21 +116,77 @@ pub fn supervise() -> io::Re let mut handler = H::default(); let mut resp_buf = alloc_seccomp_notif_resp(); + let mut task_cancel = cancel_rx.clone(); + let handler_tx = handler_tx.clone(); + spawned += 1; + + tokio::spawn(async move { + /// What one select round decided, owned so the notify borrow + /// ends before the listener is used again. + enum Step { + Respond(u64), + Cancelled, + Eof, + Fail(io::Error), + } - join_set.spawn(async move { - while let Some(notify) = listener.next().await? { - let _span = span!(Level::TRACE, "notify loop tick"); - // Errors on the supervisor side could be caused by a target process aborting. - // It shouldn't break the syscall handling loop as there might be target processes. - let _handle_result = handler.handle_notify(notify); - let req_id = notify.id; - listener.send_continue(req_id, &mut resp_buf)?; + loop { + let step = { + let next = listener.next(); + pin_mut!(next); + let cancelled = task_cancel.changed(); + pin_mut!(cancelled); + match select(cancelled, next).await { + Either::Left(_) => Step::Cancelled, + Either::Right((Ok(Some(notify)), _)) => { + let _span = span!(Level::TRACE, "notify loop tick"); + // Errors on the supervisor side could be caused by a target process aborting. + // It shouldn't break the syscall handling loop as there might be target processes. + let _handle_result = handler.handle_notify(notify); + Step::Respond(notify.id) + } + Either::Right((Ok(None), _)) => Step::Eof, + Either::Right((Err(error), _)) => Step::Fail(error), + } + }; + match step { + Step::Respond(req_id) => { + if let Err(error) = listener.send_continue(req_id, &mut resp_buf) { + let _ = handler_tx.send(Err(error)); + return; + } + } + Step::Cancelled => { + // The supervisor stopped. Hand over what was + // recorded, then keep answering notifications so + // live targets keep working. Closing the notify fd + // instead would fail their filtered syscalls. + let _ = handler_tx.send(Ok(std::mem::take(&mut handler))); + while let Ok(Some(notify)) = listener.next().await { + let req_id = notify.id; + if listener.send_continue(req_id, &mut resp_buf).is_err() { + return; + } + } + return; + } + Step::Eof => { + let _ = handler_tx.send(Ok(handler)); + return; + } + Step::Fail(error) => { + let _ = handler_tx.send(Err(error)); + return; + } + } } - io::Result::Ok(handler) }); } - let mut handlers = Vec::::new(); - while let Some(handler) = join_set.join_next().await.transpose()? { + + drop(handler_tx); + let mut handlers = Vec::::with_capacity(spawned); + for _ in 0..spawned { + let handler = handler_rx.recv().await.expect("every spawned task sends one message"); handlers.push(handler?); } Ok(handlers) diff --git a/crates/fspy_shared/Cargo.toml b/crates/fspy_shared/Cargo.toml index c854d0239..481396fd2 100644 --- a/crates/fspy_shared/Cargo.toml +++ b/crates/fspy_shared/Cargo.toml @@ -14,8 +14,6 @@ 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"] } vite_path = { workspace = true } [target.'cfg(target_os = "windows")'.dependencies] @@ -27,7 +25,6 @@ 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 diff --git a/crates/fspy_shared/src/ipc/channel/gate.rs b/crates/fspy_shared/src/ipc/channel/gate.rs new file mode 100644 index 000000000..57830934c --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/gate.rs @@ -0,0 +1,232 @@ +//! A closeable writer gate built from one shared word. +//! +//! Word layout: bit 63 is `CLOSED`, bits 0..63 count the guards currently held. +//! The word must read `0` before first use; callers get that from +//! zero-initialized memory. +//! +//! A writer takes a guard through [`Gate::enter`] and drops it when its work is +//! finished. The reader side calls [`Gate::close`] exactly once at the moment it +//! wants the memory frozen. `close` returns the number of guards that were still +//! held. Zero means every guard ever issued has already been dropped, and +//! everything written while those guards were held happens-before the `close`. +//! +//! # Why one word +//! +//! The closed bit and the count share a word on purpose. The near-counterexample +//! is: a writer reads the word, sees it open, the reader closes, and the writer +//! then increments and writes into memory the reader is already reading. That +//! cannot happen here because both halves are read-modify-write operations on +//! the same location, so they are totally ordered by that location's +//! modification order. Either the writer's compare-and-swap precedes the close, +//! and then the close's return value counts it, so the reader refuses to read; +//! or the close precedes it, and the writer's retry observes `CLOSED` and never +//! claims. A stale load never turns into a successful claim, because the +//! decision *is* the compare-and-swap. Splitting the bit and the count into two +//! atomics would reintroduce exactly that race. +//! +//! # Why `close() == 0` means frozen +//! +//! A guard is released only after the work it protects is complete, so a zero +//! count proves that every admitted claim ran to completion. The `Release` on +//! guard drop and the `Acquire` on `close` then make those writes visible to the +//! reader, not merely finished. +//! +//! A process killed between `enter` and the guard's drop leaks its count +//! permanently. That fails closed: `close` reports a nonzero count forever and +//! the caller never reads the memory. + +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Set once the gate is closed. No claim is admitted afterwards. +const CLOSED: u64 = 1 << 63; +/// The guard count occupies every bit below [`CLOSED`]. +const COUNT_MASK: u64 = CLOSED - 1; + +/// A gate over one caller-provided word. +pub(super) struct Gate<'a> { + word: &'a AtomicU64, +} + +/// Why [`Gate::enter`] refused to admit a writer. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum EnterError { + /// The gate is closed. The writer must discard its work silently. + Closed, + /// The guard count saturated. Unreachable in practice; refusing is still + /// better than wrapping into the closed bit. + Saturated, +} + +impl<'a> Gate<'a> { + /// Wraps `word`, which must read `0` before the first [`Gate::enter`] or + /// [`Gate::close`]. + pub(super) const fn new(word: &'a AtomicU64) -> Self { + Self { word } + } + + /// Admits a writer and counts it in one compare-and-swap. + /// + /// Checking the closed bit and publishing the increment are the same + /// successful operation, so there is no observable "decided to enter but not + /// yet counted" state. + pub(super) fn enter(&self) -> Result, EnterError> { + // Relaxed is enough. Entering publishes no data: the writer's stores + // come after it on the same thread, and the read side learns about + // them through the guard's release drop, not through this increment. + // Atomic read-modify-writes always see the newest value of the word, + // whatever their ordering, so the closed check cannot act on a stale + // word either. + self.word + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + if current & CLOSED != 0 || current & COUNT_MASK == COUNT_MASK { + None + } else { + Some(current + 1) + } + }) + .map(|_| GateGuard { word: self.word }) + .map_err( + |current| { + if current & CLOSED == 0 { EnterError::Saturated } else { EnterError::Closed } + }, + ) + } + + /// Closes the gate and reports how many guards were still held. + /// + /// Idempotent: each call reports the count at that moment. A return value of + /// `0` is the proof that the guarded memory is frozen and fully visible. + /// + /// Why acquire is enough even though enters are relaxed: a count of `0` + /// means the word's history balances, so the operation right before this + /// one in the word's modification order must be a guard's release drop. + /// Read-modify-writes carry release sequences forward, so this acquire + /// synchronizes with every earlier release drop, not just the newest one, + /// and everything each writer stored before dropping its guard is visible + /// from here on. + pub(super) fn close(&self) -> u64 { + self.word.fetch_or(CLOSED, Ordering::AcqRel) & COUNT_MASK + } +} + +/// A live writer's token. Dropping it makes everything written while it was +/// held visible to whoever closes the gate. +#[derive(Debug)] +pub(super) struct GateGuard<'a> { + word: &'a AtomicU64, +} + +impl Drop for GateGuard<'_> { + fn drop(&mut self) { + self.word.fetch_sub(1, Ordering::Release); + } +} + +#[cfg(test)] +mod tests { + use std::{sync::atomic::AtomicUsize, thread}; + + use super::*; + + #[test] + fn enter_and_drop_balance_the_count() { + let word = AtomicU64::new(0); + let gate = Gate::new(&word); + + let first = gate.enter().unwrap(); + let second = gate.enter().unwrap(); + assert_eq!(word.load(Ordering::Relaxed) & COUNT_MASK, 2); + + drop(first); + drop(second); + assert_eq!(gate.close(), 0); + } + + #[test] + fn enter_is_refused_after_close() { + let word = AtomicU64::new(0); + let gate = Gate::new(&word); + + assert_eq!(gate.close(), 0); + assert_eq!(gate.enter().unwrap_err(), EnterError::Closed); + // Closing again is harmless and still reports no writer in flight. + assert_eq!(gate.close(), 0); + } + + #[test] + fn close_reports_a_live_guard() { + let word = AtomicU64::new(0); + let gate = Gate::new(&word); + + let guard = gate.enter().unwrap(); + assert_eq!(gate.close(), 1); + drop(guard); + // The count drains even after the close, but the gate stays closed. + assert_eq!(gate.close(), 0); + assert_eq!(gate.enter().unwrap_err(), EnterError::Closed); + } + + #[test] + fn leaked_guard_keeps_the_count_forever() { + let word = AtomicU64::new(0); + let gate = Gate::new(&word); + + std::mem::forget(gate.enter().unwrap()); + + assert_eq!(gate.close(), 1); + assert_eq!(gate.close(), 1); + } + + #[test] + fn enter_refuses_a_saturated_count() { + let word = AtomicU64::new(COUNT_MASK); + let gate = Gate::new(&word); + + assert_eq!(gate.enter().unwrap_err(), EnterError::Saturated); + } + + /// The invariant the whole protocol rests on: whenever `close` returns 0, + /// every successful `enter`'s side effect is already visible, and every + /// later `enter` fails. + #[test] + fn close_returning_zero_means_frozen() { + const WRITERS: usize = 4; + const ROUNDS: usize = 64; + + for _ in 0..ROUNDS { + let word = AtomicU64::new(0); + // Stands in for the shared memory the gate protects. + let published = AtomicUsize::new(0); + let admitted = AtomicUsize::new(0); + + let (in_flight, published_at_close) = thread::scope(|scope| { + for _ in 0..WRITERS { + scope.spawn(|| { + let gate = Gate::new(&word); + if let Ok(guard) = gate.enter() { + admitted.fetch_add(1, Ordering::Relaxed); + // The side effect the guard protects. + published.fetch_add(1, Ordering::Relaxed); + drop(guard); + } + }); + } + let in_flight = Gate::new(&word).close(); + (in_flight, published.load(Ordering::Relaxed)) + }); + + if in_flight == 0 { + // A writer can only be admitted before the close, so every + // admitted writer had already dropped its guard, and the + // close's acquire makes its effect visible at that instant. + assert_eq!( + published_at_close, + admitted.load(Ordering::Relaxed), + "a writer was admitted but its effect was not published at close" + ); + } + // Whatever the close observed, the gate is shut for good. + assert_eq!(Gate::new(&word).enter().unwrap_err(), EnterError::Closed); + } + } +} diff --git a/crates/fspy_shared/src/ipc/channel/gated.rs b/crates/fspy_shared/src/ipc/channel/gated.rs new file mode 100644 index 000000000..3923d082f --- /dev/null +++ b/crates/fspy_shared/src/ipc/channel/gated.rs @@ -0,0 +1,455 @@ +//! The safe surface over a shared memory region: a close gate in front of a +//! frame arena. +//! +//! Region layout: +//! +//! ```text +//! | gate word: AtomicU64 | padding to 64 bytes | frame arena | +//! ``` +//! +//! The 64-byte header keeps the gate word and the arena's end-offset word on +//! separate cache lines and preserves the arena's alignment requirements. The +//! gate itself knows nothing about this layout, and neither does the channel +//! layer above: everything in between lives here. +//! +//! Writers reach the arena only through [`GatedShmWriter`], which holds a gate +//! guard for the lifetime of every frame. The read end starts as a +//! [`GatedShmReceiver`], and the only way to obtain a [`GatedShmReader`] is a +//! successful [`GatedShmReceiver::close`], so "never read memory that is still +//! being written" is a property of the types rather than of a convention. + +use std::{ + fmt, + num::NonZeroUsize, + ops::{Deref, DerefMut}, + ptr::slice_from_raw_parts_mut, + slice, + sync::{Arc, atomic::AtomicU64}, +}; + +use wincode::{SchemaWrite, config::DefaultConfig}; + +use super::{ + gate::{EnterError, Gate, GateGuard}, + shm_io::{self, AsRawSlice, ShmReader, ShmWriter}, +}; + +/// Bytes reserved in front of the arena for the gate word. +const GATE_REGION_SIZE: usize = 64; + +/// Why a frame could not be claimed. +#[derive(thiserror::Error, Debug)] +pub enum ClaimError { + /// The gate is closed. The caller must discard the access silently; this is + /// the by-design outcome for a process that writes after the traced root + /// process has exited. + #[error("the shared memory is closed for writing")] + Closed, + /// The arena is full, or the gate count saturated. Both are + /// impossible-in-practice states that callers treat as fatal. + #[error("the shared memory has no room for another frame")] + Capacity, +} + +impl From for ClaimError { + fn from(error: EnterError) -> Self { + match error { + EnterError::Closed => Self::Closed, + EnterError::Saturated => Self::Capacity, + } + } +} + +/// Why an encoded value could not be appended. +#[derive(thiserror::Error, Debug)] +pub enum WriteEncodedError { + /// The frame could not be claimed. [`ClaimError::Closed`] is the silent-drop + /// case. + #[error(transparent)] + Claim(#[from] ClaimError), + /// The value could not be encoded into the claimed frame. + #[error("failed to encode a value into shared memory")] + Encode(#[from] wincode::error::WriteError), + /// The value encodes to nothing, which the frame protocol cannot represent. + #[error("tried to write a zero-sized frame into shared memory")] + ZeroSizedFrame, +} + +/// A close found writers still in flight, so the region cannot be read. +#[derive(thiserror::Error, Debug)] +#[error("{in_flight} traced write(s) were still in flight when the channel closed")] +pub struct StillWriting { + /// How many gate guards were held at the instant of the close. + pub in_flight: u64, +} + +/// The write end of a gated region. +pub struct GatedShmWriter { + region: Arc, + arena: ShmWriter>, +} + +impl GatedShmWriter { + /// Creates the write end. + /// + /// # Safety + /// + /// These are the rules nothing but the caller can check: + /// + /// - `mem.as_raw_slice()` returns a stable, valid, 8-byte-aligned region of + /// at least `GATE_REGION_SIZE` plus the arena's own minimum, for as long + /// as this writer lives; + /// - the region was zero-initialized before any access; + /// - across *all* processes, the region is accessed exclusively through + /// [`GatedShmWriter`] and [`GatedShmReceiver`]/[`GatedShmReader`]. + /// + /// # Panics + /// + /// Panics if the region is too small or misaligned for a gate word. + pub unsafe fn new(mem: M) -> Self { + assert_gated_region(&mem); + let region = Arc::new(mem); + // SAFETY: `Arena` addresses the region past its gate header, which the + // caller's contract makes valid, stable and zero-initialized, and which + // nothing outside this protocol touches. + let arena = unsafe { ShmWriter::new(Arena { region: Arc::clone(®ion) }) }; + Self { region, arena } + } + + /// Claims a frame of `size` bytes, holding the gate open until it is + /// dropped. + /// + /// # Errors + /// + /// Returns [`ClaimError::Closed`] once the read end has closed the gate, and + /// [`ClaimError::Capacity`] when the arena cannot fit the frame. + pub fn claim_frame(&self, size: NonZeroUsize) -> Result, ClaimError> { + let gate = self.gate().enter()?; + let frame = self.arena.claim_frame(size).ok_or(ClaimError::Capacity)?; + Ok(FrameMut { frame, _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. + pub fn write_encoded>( + &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(), + }) + } + + fn gate(&self) -> Gate<'_> { + gate_of(self.region.as_ref()) + } +} + +/// 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. +pub struct FrameMut<'a> { + frame: shm_io::FrameMut<'a>, + _gate: GateGuard<'a>, +} + +impl Deref for FrameMut<'_> { + type Target = [u8]; + + fn deref(&self) -> &Self::Target { + &self.frame + } +} + +impl DerefMut for FrameMut<'_> { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.frame + } +} + +/// The read end before the region has been frozen. +/// +/// Built once, right after the mapping is created, while it is still +/// zero-initialized and no other process knows about it. +pub struct GatedShmReceiver { + region: M, +} + +impl GatedShmReceiver { + /// Creates the read end. + /// + /// # Safety + /// + /// Same rules as [`GatedShmWriter::new`]. + /// + /// # Panics + /// + /// Panics if the region is too small or misaligned for a gate word. + pub unsafe fn new(mem: M) -> Self { + assert_gated_region(&mem); + Self { region: mem } + } + + /// Closes the gate and, if nothing was in flight, hands out the reader. + /// + /// This is a single atomic operation and never blocks. On success the + /// region is never written again: every later claim fails against the + /// closed bit, the constructor's rules keep writers outside this protocol + /// away, and the release/acquire pairing makes every past write visible. + /// That is what makes borrowing it as `&[u8]` sound. + /// + /// # Errors + /// + /// Returns [`StillWriting`] when a guard was held at the instant of the + /// close. No reader exists in that case, by construction. + pub fn close(self) -> Result, StillWriting> { + let in_flight = gate_of(&self.region).close(); + if in_flight == 0 { + Ok(GatedShmReader { frames: ShmReader::new(FrozenArena { region: self.region }) }) + } else { + Err(StillWriting { in_flight }) + } + } +} + +/// A frozen region, obtainable only from a successful [`GatedShmReceiver::close`]. +pub struct GatedShmReader { + frames: ShmReader>, +} + +impl GatedShmReader { + /// Iterates over every frame written before the close. + pub fn iter_frames(&self) -> impl Iterator { + self.frames.iter_frames() + } +} + +impl fmt::Debug for GatedShmReader { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.debug_struct("GatedShmReader").finish_non_exhaustive() + } +} + +/// Addresses the arena that follows a region's gate header. +struct Arena { + region: Arc, +} + +impl AsRawSlice for Arena { + fn as_raw_slice(&self) -> *mut [u8] { + arena_of(self.region.as_raw_slice()) + } +} + +/// Addresses the arena of a region that is known to be frozen. +struct FrozenArena { + region: M, +} + +impl AsRef<[u8]> for FrozenArena { + fn as_ref(&self) -> &[u8] { + let arena = arena_of(self.region.as_raw_slice()); + // SAFETY: this type only exists after `GatedShmReceiver::close` reported + // nothing in flight, so the region is frozen for the whole lifetime of + // this value, and the constructor's rules keep it valid and free of + // writers outside the protocol. + unsafe { slice::from_raw_parts(arena.cast::(), arena.len()) } + } +} + +/// Borrows a region's gate word. +#[expect( + clippy::cast_ptr_alignment, + reason = "the constructors assert the region is aligned for the gate word" +)] +fn gate_of(region: &M) -> Gate<'_> { + let base = region.as_raw_slice().cast::(); + // SAFETY: the constructors checked the region's size and 8-byte alignment, + // the region outlives this borrow, and the caller's rules guarantee the word + // started at zero and is only ever used as this gate. + Gate::new(unsafe { AtomicU64::from_ptr(base) }) +} + +/// Splits the arena off a region's gate header. +fn arena_of(region: *mut [u8]) -> *mut [u8] { + let len = region.len(); + debug_assert!(len > GATE_REGION_SIZE, "gated region is smaller than its gate header"); + // SAFETY: the constructors checked that the region is longer than its gate + // header, so this stays inside the same allocation. + let arena = unsafe { region.cast::().add(GATE_REGION_SIZE) }; + slice_from_raw_parts_mut(arena, len - GATE_REGION_SIZE) +} + +/// Checks the parts of the constructors' contract that are cheap to check. +fn assert_gated_region(mem: &M) { + let region = mem.as_raw_slice(); + assert!( + region.len() > GATE_REGION_SIZE, + "a gated region must be larger than its {GATE_REGION_SIZE}-byte gate header" + ); + assert_eq!( + region.cast::() as usize % align_of::(), + 0, + "a gated region must be aligned for its gate word" + ); +} + +#[cfg(test)] +mod tests { + use std::{str::from_utf8, sync::Arc, thread}; + + use super::*; + + /// A plain heap region standing in for shared memory, so these tests also + /// run under miri. + #[derive(Clone)] + struct MockRegion { + // `u64` elements give the region the alignment a gate word needs. + mem: Arc>, + len: usize, + } + + // SAFETY: the backing allocation is shared through an `Arc`, and every access + // goes through the gated protocol, which is what makes concurrent use sound. + unsafe impl Send for MockRegion {} + // SAFETY: as above. + unsafe impl Sync for MockRegion {} + + impl MockRegion { + fn alloc(len: usize) -> Self { + Self { mem: Arc::new(vec![0u64; len / size_of::() + 1]), len } + } + } + + impl AsRawSlice for MockRegion { + fn as_raw_slice(&self) -> *mut [u8] { + slice_from_raw_parts_mut(Vec::as_ptr(&self.mem).cast::().cast_mut(), self.len) + } + } + + fn endpoints(len: usize) -> (GatedShmWriter, GatedShmReceiver) { + let region = MockRegion::alloc(len); + // SAFETY: `MockRegion` is a valid, stable, 8-byte-aligned, + // zero-initialized allocation used only through these two types. + unsafe { (GatedShmWriter::new(region.clone()), GatedShmReceiver::new(region)) } + } + + fn write_frame(writer: &GatedShmWriter, bytes: &[u8]) -> Result<(), ClaimError> { + let mut frame = writer.claim_frame(NonZeroUsize::new(bytes.len()).unwrap())?; + frame.copy_from_slice(bytes); + Ok(()) + } + + #[test] + fn write_close_read_roundtrip() { + let (writer, receiver) = endpoints(1024); + + write_frame(&writer, b"hello").unwrap(); + write_frame(&writer, b"world").unwrap(); + + let reader = receiver.close().unwrap(); + let frames: Vec<&[u8]> = reader.iter_frames().collect(); + assert_eq!(frames, vec![b"hello".as_slice(), b"world".as_slice()]); + } + + #[test] + fn close_with_a_live_frame_reports_still_writing() { + let (writer, receiver) = endpoints(1024); + + let frame = writer.claim_frame(NonZeroUsize::new(4).unwrap()).unwrap(); + // Reading anyway is not expressible: `close` consumes the receiver and + // only hands back a reader on success, so this error value is the end of + // the road for the read side. + let error = receiver.close().unwrap_err(); + + assert_eq!(error.in_flight, 1); + drop(frame); + } + + #[test] + fn claims_after_close_are_refused() { + let (writer, receiver) = endpoints(1024); + + write_frame(&writer, b"before").unwrap(); + let reader = receiver.close().unwrap(); + + assert!(matches!(write_frame(&writer, b"after"), Err(ClaimError::Closed))); + assert!(matches!( + writer.write_encoded(&7u32), + Err(WriteEncodedError::Claim(ClaimError::Closed)) + )); + + let frames: Vec<&[u8]> = reader.iter_frames().collect(); + assert_eq!(frames, vec![b"before".as_slice()]); + } + + #[test] + fn a_full_arena_reports_capacity() { + let (writer, receiver) = endpoints(GATE_REGION_SIZE + 64); + + while write_frame(&writer, b"filler").is_ok() {} + assert!(matches!(write_frame(&writer, b"filler"), Err(ClaimError::Capacity))); + + // Whatever fitted before the arena filled up is still readable. + let reader = receiver.close().unwrap(); + assert!(reader.iter_frames().all(|frame| frame == b"filler")); + } + + #[test] + fn write_encoded_roundtrips_through_the_gate() { + let (writer, receiver) = endpoints(1024); + + writer.write_encoded(&4242u32).unwrap(); + + let reader = receiver.close().unwrap(); + let frame = reader.iter_frames().next().unwrap(); + assert_eq!(wincode::deserialize_exact::(frame).unwrap(), 4242); + } + + #[test] + fn concurrent_writers_are_all_published_by_a_successful_close() { + const WRITERS: usize = 4; + const PER_WRITER: usize = 20; + + let region = MockRegion::alloc(64 * 1024); + // SAFETY: as in `endpoints`. + let receiver = unsafe { GatedShmReceiver::new(region.clone()) }; + + thread::scope(|scope| { + for writer_index in 0..WRITERS { + let region = region.clone(); + scope.spawn(move || { + // SAFETY: as in `endpoints`. + let writer = unsafe { GatedShmWriter::new(region) }; + for frame_index in 0..PER_WRITER { + write_frame(&writer, format!("{writer_index} {frame_index}").as_bytes()) + .unwrap(); + } + }); + } + }); + + // Every writer joined before the close, so it must report a clean gate, + // and that `Ok` is also the synchronization that makes their frames + // visible here. + let reader = receiver.close().unwrap(); + let mut seen: Vec<&str> = + reader.iter_frames().map(|frame| from_utf8(frame).unwrap()).collect(); + seen.sort_unstable(); + + assert_eq!(seen.len(), WRITERS * PER_WRITER); + assert_eq!(seen[0], "0 0"); + } +} diff --git a/crates/fspy_shared/src/ipc/channel/mod.rs b/crates/fspy_shared/src/ipc/channel/mod.rs index fb1b2ef14..61ea485d7 100644 --- a/crates/fspy_shared/src/ipc/channel/mod.rs +++ b/crates/fspy_shared/src/ipc/channel/mod.rs @@ -1,218 +1,198 @@ //! Fast mpsc IPC channel implementation based on shared memory. - +//! +//! The shared memory carries its own close gate: a writer holds a gate guard +//! while it writes a frame, and the receiver closes the gate with one atomic +//! operation. The closed flag and the writer count live in the same atomic +//! word, so a process cannot write without being counted. The file lock this +//! replaces got that wrong: a tool that closed file descriptors it did not +//! recognize released the lock while it kept write access to the mapping. +//! +//! The close does not wait for anyone. Accesses made after the traced root +//! process exits are dropped by design, and a run that still had a write in +//! flight at that instant is reported as incomplete rather than read. + +mod gate; +mod gated; mod shm_io; -use std::{env::temp_dir, fs::File, io, ops::Deref, path::PathBuf}; +use std::io; use fspy_shm::{Mapping, ShmKeeper}; -pub use shm_io::FrameMut; -use shm_io::{ShmReader, ShmWriter}; -use tracing::debug; -use uuid::Uuid; +pub use gated::{ + ClaimError, FrameMut, GatedShmReader, GatedShmReceiver, GatedShmWriter, + StillWriting as TrackingIncomplete, WriteEncodedError, +}; use wincode::{SchemaRead, SchemaWrite}; -use super::NativeStr; +use super::{NativeStr, PathAccess}; /// Serializable configuration to create channel senders. #[derive(SchemaWrite, SchemaRead, Clone, Debug)] pub struct ChannelConf { - lock_file_path: Box, shm_id: Box, } +/// The write end of an IPC channel. Every frame is claimed through the gate, so +/// no writer can bypass it. +pub type Sender = GatedShmWriter; + +/// The unique read end of an IPC channel, before it is closed. +/// +/// Holds the shared memory's keeper: dropping the receiver, or closing it, +/// removes the backing file's name, after which no new sender can be created. +pub struct Receiver { + keeper: ShmKeeper, + inner: GatedShmReceiver, +} + +impl Receiver { + /// Closes the gate and freezes the region. + /// + /// Also drops the keeper: the attach window ends here, so a process that + /// opens the channel later fails at `open` instead of at its first claim. + /// + /// # Errors + /// + /// Returns [`TrackingIncomplete`] when a write was still in flight at the + /// instant of the close. The region is not read in that case. + pub fn close(self) -> Result { + drop(self.keeper); + self.inner.close() + } +} + +/// A frozen channel, handed out by [`Receiver::close`]. +pub type ChannelFrames = GatedShmReader; + /// 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 { shm_id: keeper.id().into() }; - 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)) + // SAFETY: the backing file was just created zero-initialized, its identifier + // is not yet published to any other process, the mapping stays valid while + // the receiver holds it, and `Sender` is the only other way to reach it. + let inner = unsafe { GatedShmReceiver::new(mapping) }; + Ok((conf, Receiver { keeper, inner })) } 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. + /// Fails once the receiver has been closed or dropped, because both remove + /// the backing file's name. A sender created before that keeps working, but + /// its claims fail with [`ClaimError::Closed`] after the close. #[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 }) + // SAFETY: the mapping was created by `channel` under the same contract, + // it stays valid while this sender holds it, and every process reaches it + // only through `Sender` or `Receiver`. + Ok(unsafe { Sender::new(mapping) }) } } -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 +impl ChannelFrames { + /// Decodes every path access recorded before the close. + /// + /// # Panics + /// + /// Panics if a frame does not decode. The frames were written by this same + /// binary against this same schema, so that is a bug worth hearing about + /// rather than a recoverable condition. + pub fn iter_path_accesses(&self) -> impl Iterator> { + // The frames were written by this same binary, so a decode failure is a + // bug worth hearing about rather than a recoverable condition. + self.iter_frames().map(|frame| wincode::deserialize_exact(frame).unwrap()) } } #[cfg(test)] mod tests { - use std::{num::NonZeroUsize, str::from_utf8}; + use std::{ + ffi::OsString, + io::{BufRead, BufReader, Read, Write}, + num::NonZeroUsize, + process::{Command, Stdio}, + 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(); + /// Smaller than a page, but large enough for the gate header plus a few + /// frames. + const SMALL_CAPACITY: usize = 256; + /// Frames the straggler grandchild writes after its parent has exited. + const STRAGGLER_FRAMES: usize = 16; + /// Marks the re-executed process as the straggler grandchild. + const STRAGGLER_ENV: &str = "FSPY_IPC_TEST_STRAGGLER"; + /// Carries the child's own command line so it can re-execute itself. + /// + /// `subprocess_test` dispatches from a library constructor, and on Linux + /// `std::env::args_os` is empty that early, so the arguments travel through + /// the environment instead. + const STRAGGLER_ARGV_ENV: &str = "FSPY_IPC_TEST_STRAGGLER_ARGV"; + + #[test] + fn smoke() { + let (conf, receiver) = channel(SMALL_CAPACITY).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()); + assert!(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]); + let frames = receiver.close().unwrap(); + let mut frames = frames.iter_frames(); + assert_eq!(frames.next().unwrap(), &[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(); + #[test] + fn claims_refused_after_close() { + let (conf, receiver) = channel(SMALL_CAPACITY).unwrap(); + // A sender that attached while the channel was open. + let sender = conf.sender().unwrap(); + let _frames = receiver.close().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")); + // Its claims are refused after the close. + let claim = sender.claim_frame(NonZeroUsize::new(2).unwrap()); + assert!(matches!(claim, Err(ClaimError::Closed))); + + // The close also dropped the keeper, so a process that starts later + // does not even reach the shared memory. + assert!(conf.sender().is_err()); } - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[test] #[expect(clippy::print_stdout, reason = "test diagnostics")] - async fn forbid_new_senders_after_receiver_dropped() { - let (conf, receiver) = channel(42).unwrap(); + fn forbid_new_senders_after_receiver_dropped() { + let (conf, receiver) = channel(SMALL_CAPACITY).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(); + let output = Command::from(cmd).output().unwrap(); assert_eq!(B(&output.stdout), B("false")); } - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn concurrent_senders() { + #[test] + 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)| { @@ -223,7 +203,7 @@ mod tests { .unwrap() .copy_from_slice(data_to_send.as_bytes()); }); - let output = std::process::Command::from(cmd).output().unwrap(); + let output = Command::from(cmd).output().unwrap(); assert!( output.status.success(), "Failed to send in iteration {}: {:?}", @@ -231,12 +211,102 @@ mod tests { B(&output.stderr) ); } - let lock = receiver.lock().unwrap(); - let mut received_values: Vec = lock + let frames = receiver.close().unwrap(); + let mut received_values: Vec = frames .iter_frames() .map(|frame| from_utf8(frame).unwrap().parse::().unwrap()) .collect(); received_values.sort_unstable(); assert_eq!(received_values, (0u16..200).collect::>()); } + + /// The topology behind the descriptor-sweep bug: a child spawns a detached + /// grandchild and exits, and the grandchild keeps writing. Closing must + /// collect what the grandchild already wrote and must not wait for it to go + /// away. + #[test] + fn straggler_writes_through_close() { + let (conf, receiver) = channel(8192).unwrap(); + let cmd = command_for_fn!(conf, |conf: ChannelConf| { + if std::env::var_os(STRAGGLER_ENV).is_none() { + // The child's only job is to leave a detached descendant behind + // and exit. Re-executing itself keeps both roles in one closure. + let argv = std::env::var(STRAGGLER_ARGV_ENV).unwrap(); + let mut grandchild = Command::new(std::env::current_exe().unwrap()); + grandchild.args(argv.split('\n')).env(STRAGGLER_ENV, "1"); + #[expect( + clippy::zombie_processes, + reason = "leaving the descendant behind unwaited is the situation under test" + )] + grandchild.spawn().unwrap(); + return; + } + + let sender = conf.sender().unwrap(); + for i in 0..STRAGGLER_FRAMES { + let data = i.to_string(); + sender + .claim_frame(NonZeroUsize::new(data.len()).unwrap()) + .unwrap() + .copy_from_slice(data.as_bytes()); + } + // Announce that every frame is written, then linger on the inherited + // stdin until the test closes the channel and releases us. + let mut stdout = std::io::stdout(); + stdout.write_all(b"written\n").unwrap(); + stdout.flush().unwrap(); + let mut ignored = Vec::new(); + std::io::stdin().read_to_end(&mut ignored).unwrap(); + }); + + let mut cmd = cmd; + let argv = cmd + .args + .iter() + .map(|arg| arg.to_str().unwrap().to_owned()) + .collect::>() + .join("\n"); + cmd.envs.insert(OsString::from(STRAGGLER_ARGV_ENV), OsString::from(argv)); + + let mut child = + Command::from(cmd).stdin(Stdio::piped()).stdout(Stdio::piped()).spawn().unwrap(); + let stdin = child.stdin.take().unwrap(); + let mut stdout = BufReader::new(child.stdout.take().unwrap()); + assert!(child.wait().unwrap().success()); + + // The child is gone; the grandchild is still alive and holds both pipes. + let mut ready = String::new(); + stdout.read_line(&mut ready).unwrap(); + assert_eq!(ready.trim_end(), "written"); + + // Closing does not wait for the grandchild's lifetime, and everything it + // wrote before the close is there. + let frames = receiver.close().unwrap(); + let mut written: Vec = frames + .iter_frames() + .map(|frame| from_utf8(frame).unwrap().parse::().unwrap()) + .collect(); + written.sort_unstable(); + assert_eq!(written, (0..STRAGGLER_FRAMES).collect::>()); + + // Closing the pipe lets the grandchild exit. + drop(stdin); + } + + /// A process killed between claiming a frame and finishing it leaks its gate + /// count, which must fail closed rather than expose half-written memory. + #[test] + fn leaked_gate_guard_reports_incomplete() { + let (conf, receiver) = channel(SMALL_CAPACITY).unwrap(); + let cmd = command_for_fn!(conf, |conf: ChannelConf| { + let sender = conf.sender().unwrap(); + let frame = sender.claim_frame(NonZeroUsize::new(2).unwrap()).unwrap(); + // Stands in for a process dying mid-frame. + std::mem::forget(frame); + }); + assert!(Command::from(cmd).status().unwrap().success()); + + let error = receiver.close().unwrap_err(); + assert_eq!(error.in_flight, 1); + } } diff --git a/crates/fspy_shm/README.md b/crates/fspy_shm/README.md index 6be9c9011..d63fb9356 100644 --- a/crates/fspy_shm/README.md +++ b/crates/fspy_shm/README.md @@ -18,7 +18,7 @@ The public API is defined in [`src/lib.rs`](src/lib.rs). | `Mapping::as_ptr()` | Returns a mutable raw pointer to the first byte. | | `Mapping::as_slice()` | Returns the bytes as a shared slice. The caller must prevent mutation for its lifetime. | -`ShmKeeper` is the name: while it lives, `open` succeeds, and dropping it removes the backing file. `ShmHandle` is the opened file: `create` returns one so the creator never looks its own file up by name, and `open` returns one to everybody else. `Mapping` is the bytes: it keeps them alive until dropped and can do nothing else. None of the three synchronizes memory access. The fspy channel adds that on top with atomic frame headers and a lock file: senders hold a shared file lock while writing, and the receiver takes the exclusive lock before reading, which waits for existing senders and rejects new ones. +`ShmKeeper` is the name: while it lives, `open` succeeds, and dropping it removes the backing file. `ShmHandle` is the opened file: `create` returns one so the creator never looks its own file up by name, and `open` returns one to everybody else. `Mapping` is the bytes: it keeps them alive until dropped and can do nothing else. None of the three synchronizes memory access. The fspy channel adds that on top with atomic frame headers and a close gate stored in the mapping itself: every writer holds a gate guard while it writes, and the receiver closes the gate with one atomic operation, which refuses every later writer and reports whether any write was still in flight. Every byte in a mapping returned by `create` is initially zero. `open` exposes the mapping's current contents and does not reinitialize them. @@ -67,6 +67,6 @@ Earlier revisions rejected temporary files because dirty pages can reach disk. O - Dropping the keeper removes the backing file's name, so later opens fail. This is cleanup, not a stop signal: processes that already opened the shared memory keep reading and writing. The fspy channel stops writers with the close gate it stores in the shared bytes. - An `ShmHandle` and its `Mapping`s stay usable after the keeper is gone. They keep the bytes alive and cannot extend the identifier's validity. -The channel guards the same window from its own side: [`ChannelConf::sender`](../fspy_shared/src/ipc/channel/mod.rs) opens and locks the receiver's exact lock-file path before it calls `fspy_shm::open`, and the receiver removes that path before dropping the keeper, so a sender that starts later fails before opening shared memory. +The channel builds on that: [`ChannelConf::sender`](../fspy_shared/src/ipc/channel/mod.rs) is just an `open`, and the receiver drops the keeper when it closes the channel. A sender that attached earlier keeps its mapping, but the close gate refuses its claims from then on. If the keeper's process is killed, its `Drop` never runs and the file stays behind: on Unix for the system's temporary-file reaper, on Windows until a cleanup tool runs. The file costs about as much disk as the run wrote into it. diff --git a/crates/vite_task/src/session/event.rs b/crates/vite_task/src/session/event.rs index 3cdfb8737..4516f0618 100644 --- a/crates/vite_task/src/session/event.rs +++ b/crates/vite_task/src/session/event.rs @@ -87,6 +87,10 @@ pub enum CacheNotUpdatedReason { /// First path that was both read and written during execution. path: RelativePathBuf, }, + /// A traced process was still writing when file tracking closed at task + /// exit, so the recorded accesses are an incomplete picture of the run and + /// caching it could miss an input or an output. + TrackingIncomplete, /// fspy isn't compiled in on this build and the task requires fspy /// (its `input` config includes auto-inference). Task ran but cannot /// be cached without tracked path accesses. diff --git a/crates/vite_task/src/session/execute/cache_update.rs b/crates/vite_task/src/session/execute/cache_update.rs index 0cf4df1c5..627f08a20 100644 --- a/crates/vite_task/src/session/execute/cache_update.rs +++ b/crates/vite_task/src/session/execute/cache_update.rs @@ -34,6 +34,26 @@ struct TrackingOutcome { /// First path that was both read and written during execution, if any. /// A non-empty value means caching this task is unsound. read_write_overlap: Option, + /// Tracking was cut short: a traced process was still writing when the + /// channel closed at task exit, so accesses may be missing. + incomplete: bool, +} + +/// The reasons a run must not be cached that follow from what fspy observed, +/// in priority order. +fn tracking_guard(tracking: &TrackingOutcome) -> Option { + if tracking.incomplete { + // We cannot tell which accesses were lost, so any cache entry built + // from this run risks a stale hit later. + return Some(CacheNotUpdatedReason::TrackingIncomplete); + } + + // fspy-inferred read-write overlap: the task wrote to a file it also read, + // so the prerun input hashes are stale and caching is unsound. (We only + // check fspy-inferred reads, not globbed_inputs. A task that writes to a + // glob-matched file without reading it produces perpetual cache misses but + // not a correctness bug.) + tracking.read_write_overlap.clone().map(|path| CacheNotUpdatedReason::InputModified { path }) } type TrackedEnvValues = BTreeMap>; @@ -98,18 +118,8 @@ pub(super) async fn update_cache( workspace_root, ); - if let Some(TrackingOutcome { read_write_overlap: Some(path), .. }) = &fspy_outcome { - // fspy-inferred read-write overlap: the task wrote to a file it also - // read, so the prerun input hashes are stale and caching is unsound. - // (We only check fspy-inferred reads, not globbed_inputs. A task that - // writes to a glob-matched file without reading it produces perpetual - // cache misses but not a correctness bug.) - return ( - CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::InputModified { - path: path.clone(), - }), - None, - ); + if let Some(reason) = fspy_outcome.as_ref().and_then(tracking_guard) { + return (CacheUpdateStatus::NotUpdated(reason), None); } if fspy_outcome.is_none() && fspy.is_some() { @@ -249,6 +259,7 @@ fn observe_fspy( path_reads: filtered_path_reads, path_writes: filtered_path_writes, read_write_overlap, + incomplete: raw.is_incomplete(), } }) } @@ -410,7 +421,44 @@ mod tests { use rustc_hash::FxHashSet; use vite_path::{AbsolutePath, RelativePathBuf}; - use super::normalize_ignored_paths; + use super::{TrackingOutcome, normalize_ignored_paths, tracking_guard}; + use crate::{collections::HashMap, session::event::CacheNotUpdatedReason}; + + fn tracking_outcome(read_write_overlap: Option<&str>, incomplete: bool) -> TrackingOutcome { + TrackingOutcome { + path_reads: HashMap::default(), + path_writes: FxHashSet::default(), + read_write_overlap: read_write_overlap.map(|path| RelativePathBuf::new(path).unwrap()), + incomplete, + } + } + + #[test] + fn complete_tracking_without_overlap_allows_caching() { + assert!(tracking_guard(&tracking_outcome(None, false)).is_none()); + } + + #[test] + fn read_write_overlap_blocks_caching() { + let reason = tracking_guard(&tracking_outcome(Some("src/generated.ts"), false)); + + let Some(CacheNotUpdatedReason::InputModified { path }) = reason else { + panic!("expected an input-modified reason, got {reason:?}"); + }; + assert_eq!(path.as_str(), "src/generated.ts"); + } + + #[test] + fn incomplete_tracking_blocks_caching_ahead_of_any_overlap() { + // Incomplete tracking wins: with accesses missing we cannot even trust + // the overlap verdict. + let reason = tracking_guard(&tracking_outcome(Some("src/generated.ts"), true)); + + assert!( + matches!(reason, Some(CacheNotUpdatedReason::TrackingIncomplete)), + "expected tracking to be reported as incomplete, got {reason:?}" + ); + } #[test] fn normalize_ignored_paths_cleans_relative_components() { diff --git a/crates/vite_task/src/session/reporter/summary.rs b/crates/vite_task/src/session/reporter/summary.rs index b6ce11848..6f564f9aa 100644 --- a/crates/vite_task/src/session/reporter/summary.rs +++ b/crates/vite_task/src/session/reporter/summary.rs @@ -114,6 +114,11 @@ pub enum SpawnOutcome { /// Set when a runner-aware tool called `disableCache()`, skipping /// cache update. tool_disabled_cache: bool, + /// `true` when a traced process was still writing as the task exited, + /// so file tracking could not produce a complete record of the run. + /// Task ran successfully but cache was not updated. + #[serde(default)] + tracking_incomplete: bool, }, /// Process exited with non-zero status. @@ -345,6 +350,10 @@ impl TaskResult { cache_update_status, CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::ToolRequested) ); + let tracking_incomplete = matches!( + cache_update_status, + CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::TrackingIncomplete) + ); match cache_status { CacheStatus::Hit { replayed_duration } => { @@ -360,6 +369,7 @@ impl TaskResult { fspy_unsupported, ipc_server_error, tool_disabled_cache, + tracking_incomplete, ), }, CacheStatus::Miss(cache_miss) => Self::Spawned { @@ -373,6 +383,7 @@ impl TaskResult { fspy_unsupported, ipc_server_error, tool_disabled_cache, + tracking_incomplete, ), }, } @@ -387,6 +398,7 @@ fn spawn_outcome_from_execution( fspy_unsupported: bool, ipc_server_error: Option, tool_disabled_cache: bool, + tracking_incomplete: bool, ) -> SpawnOutcome { match (exit_status, saved_error) { // Spawn error — process never ran @@ -398,6 +410,7 @@ fn spawn_outcome_from_execution( fspy_unsupported, ipc_server_error, tool_disabled_cache, + tracking_incomplete, }, // Process exited with non-zero code (Some(status), _) => { @@ -418,6 +431,7 @@ fn spawn_outcome_from_execution( fspy_unsupported: false, ipc_server_error: None, tool_disabled_cache: false, + tracking_incomplete: false, }, } } @@ -556,6 +570,17 @@ impl TaskResult { { return vite_str::format!("→ Not cached: read and wrote '{path}'"); } + // Tracking that could not be frozen at task exit — same overrides + // precedence as above. + if let Self::Spawned { + outcome: SpawnOutcome::Success { tracking_incomplete: true, .. }, + .. + } = self + { + return Str::from( + "→ Not cached: file tracking incomplete (a traced process was still writing at task exit)", + ); + } // fspy-unsupported-on-this-OS message — same overrides precedence as above if let Self::Spawned { outcome: SpawnOutcome::Success { fspy_unsupported: true, .. }, ..