From 82f986fcbdac6f145e9f355c3650f319afc936c8 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Fri, 24 Jul 2026 00:38:14 +0800 Subject: [PATCH] fix(pipe_socket): keep a reader open across every client FIFO write Co-authored-by: Claude Fable 5 --- CHANGELOG.md | 1 + Cargo.lock | 17 +- Cargo.toml | 1 + crates/pipe_socket/Cargo.toml | 33 ++ crates/pipe_socket/README.md | 16 + crates/pipe_socket/src/lib.rs | 118 ++++++ crates/pipe_socket/src/unix.rs | 340 ++++++++++++++++++ crates/pipe_socket/src/windows.rs | 133 +++++++ crates/pipe_socket/tests/integration.rs | 175 +++++++++ .../fspy_under_anthropic_sandbox_runtime.md | 10 +- .../snapshots/fspy_under_codex_sandbox.md | 10 +- crates/vite_task_client/Cargo.toml | 4 +- crates/vite_task_client/src/lib.rs | 60 +--- crates/vite_task_server/Cargo.toml | 7 +- crates/vite_task_server/src/lib.rs | 93 +---- crates/vite_task_server/tests/integration.rs | 14 +- 16 files changed, 853 insertions(+), 179 deletions(-) create mode 100644 crates/pipe_socket/Cargo.toml create mode 100644 crates/pipe_socket/README.md create mode 100644 crates/pipe_socket/src/lib.rs create mode 100644 crates/pipe_socket/src/unix.rs create mode 100644 crates/pipe_socket/src/windows.rs create mode 100644 crates/pipe_socket/tests/integration.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 91b6764ef..a58ad9e8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # Changelog +- **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)). - **Fixed** The task cache now supports much larger automatically tracked input sets without hitting wincode's default 4 MiB sequence preallocation limit ([#554](https://github.com/voidzero-dev/vite-task/pull/554)). diff --git a/Cargo.lock b/Cargo.lock index 110974297..b6ef0549b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2703,6 +2703,18 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "pipe_socket" +version = "0.0.0" +dependencies = [ + "nix 0.31.2", + "tempfile", + "tokio", + "uuid", + "vite_path", + "winapi", +] + [[package]] name = "pkg-config" version = "0.3.32" @@ -4283,10 +4295,10 @@ name = "vite_task_client" version = "0.0.0" dependencies = [ "native_str", + "pipe_socket", "rustc-hash", "vite_path", "vite_task_ipc_shared", - "winapi", "wincode", ] @@ -4377,13 +4389,12 @@ version = "0.0.0" dependencies = [ "futures", "native_str", + "pipe_socket", "rustc-hash", - "tempfile", "thiserror 2.0.18", "tokio", "tokio-util", "tracing", - "uuid", "vite_glob", "vite_path", "vite_task_client", diff --git a/Cargo.toml b/Cargo.toml index 1e12e8b92..2a78ecd91 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -149,6 +149,7 @@ uuid = "1.18.1" vec1 = "1.12.1" vite_glob = { path = "crates/vite_glob" } vite_graph_ser = { path = "crates/vite_graph_ser" } +pipe_socket = { path = "crates/pipe_socket" } vite_path = { path = "crates/vite_path" } vite_powershell = { path = "crates/vite_powershell" } vite_select = { path = "crates/vite_select" } diff --git a/crates/pipe_socket/Cargo.toml b/crates/pipe_socket/Cargo.toml new file mode 100644 index 000000000..4dc51e39c --- /dev/null +++ b/crates/pipe_socket/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "pipe_socket" +version = "0.0.0" +edition.workspace = true +license.workspace = true +publish = false +rust-version.workspace = true + +[dependencies] +tokio = { workspace = true, features = ["io-util", "net"] } + +[target.'cfg(unix)'.dependencies] +nix = { workspace = true, features = ["fs", "poll"] } +tempfile = { workspace = true } +uuid = { workspace = true, features = ["v4"] } +vite_path = { workspace = true } + +[target.'cfg(windows)'.dependencies] +uuid = { workspace = true, features = ["v4"] } +winapi = { workspace = true, features = ["namedpipeapi"] } + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt", "time"] } + +[target.'cfg(unix)'.dev-dependencies] +nix = { workspace = true, features = ["fs"] } + +[lints] +workspace = true + +[lib] +doctest = false +test = false diff --git a/crates/pipe_socket/README.md b/crates/pipe_socket/README.md new file mode 100644 index 000000000..7e2bf154d --- /dev/null +++ b/crates/pipe_socket/README.md @@ -0,0 +1,16 @@ +# `pipe_socket` + +Socket-style server-client IPC, implemented on named pipes rather than Unix +domain sockets. + +A server binds under an opaque name and hands that name to its clients, for +example through an environment variable. Clients connect synchronously and get +a byte stream; the server accepts each connection asynchronously with Tokio. +The API is four items: `Server::bind`, `Server::name`, `Server::accept`, and +`Client::connect`. + +The transport is named pipes on every platform: FIFOs on Unix, named pipe +objects on Windows. Both are available in environments that block Unix domain +sockets, such as coding-agent sandboxes. + +A client whose server is gone gets an error, not a hang, on both platforms. diff --git a/crates/pipe_socket/src/lib.rs b/crates/pipe_socket/src/lib.rs new file mode 100644 index 000000000..f43a5b90f --- /dev/null +++ b/crates/pipe_socket/src/lib.rs @@ -0,0 +1,118 @@ +#![doc = include_str!("../README.md")] + +use std::{ + ffi::OsStr, + io::{self, Read, Write}, + pin::Pin, + task::{Context, Poll}, +}; + +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +#[cfg(unix)] +mod unix; +#[cfg(unix)] +use unix as imp; +#[cfg(windows)] +mod windows; +#[cfg(windows)] +use windows as imp; + +#[cfg(not(any(unix, windows)))] +compile_error!("pipe_socket supports only Unix and Windows"); + +/// A named server that asynchronously accepts byte-stream connections. +pub struct Server { + inner: imp::Server, +} + +impl Server { + /// Creates a server with a new unique name. + /// + /// # Errors + /// + /// Returns an error if the platform transport cannot be created. + pub fn bind() -> io::Result { + imp::Server::bind().map(|inner| Self { inner }) + } + + /// Returns the opaque name clients use to connect to this server. + #[must_use] + pub fn name(&self) -> &OsStr { + self.inner.name() + } + + /// Waits for and accepts the next client connection. + /// + /// # Errors + /// + /// Returns an error if the connection cannot be accepted. + pub async fn accept(&mut self) -> io::Result { + self.inner.accept().await.map(|inner| ServerConnection { inner }) + } +} + +/// The server side of an accepted byte-stream connection. +pub struct ServerConnection { + inner: imp::ServerConnection, +} + +impl AsyncRead for ServerConnection { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl AsyncWrite for ServerConnection { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + +/// A synchronous client byte stream connected by a server name. +pub struct Client { + inner: imp::Client, +} + +impl Client { + /// Connects to the server identified by `name`. + /// + /// # Errors + /// + /// Returns an error if the name is invalid or the server cannot be reached. + pub fn connect(name: &OsStr) -> io::Result { + imp::Client::connect(name).map(|inner| Self { inner }) + } +} + +impl Read for Client { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.inner.read(buf) + } +} + +impl Write for Client { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.inner.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} diff --git a/crates/pipe_socket/src/unix.rs b/crates/pipe_socket/src/unix.rs new file mode 100644 index 000000000..7e51b9a4f --- /dev/null +++ b/crates/pipe_socket/src/unix.rs @@ -0,0 +1,340 @@ +//! FIFO transport. +//! +//! Coding-agent sandboxes block Unix domain sockets, so the Unix transport is +//! built from named FIFOs, which are plain files that the sandboxes allow. +//! +//! The server owns a private `0700` temp directory. It holds one well-known +//! FIFO, `connect`, that the server reads for its whole life. A connection +//! works like this: +//! +//! 1. The client creates a FIFO pair, `.request` and `.response`, +//! inside the server's directory. +//! 2. The client writes its 16-byte id to `connect`. The write is smaller +//! than `PIPE_BUF`, so concurrent clients cannot interleave. +//! 3. The server reads the id, opens the pair's far ends, and writes one +//! ready byte into the response FIFO. If the client died in the meantime, +//! these steps fail and the server moves on to the next announcement. +//! 4. The client sees the ready byte and the connection is up: requests flow +//! through one FIFO, responses through the other. +//! +//! FIFO opens block until the other end exists, and the handshake never +//! waits on that. The client opens the request FIFO read-write: POSIX +//! leaves read-write FIFO opens undefined, but Linux (see fifo(7)) and +//! macOS treat them as plain opens that never block. The response FIFO must +//! not get the same treatment: the client has to see end of file when the +//! server dies, which only happens if the client holds no write end of it, +//! so its read end opens nonblocking instead and stays for the whole +//! connection. The client also never waits on the server blindly: the +//! rendezvous open is nonblocking, and the wait for the ready byte watches +//! the rendezvous write end, which reports an error as soon as the server's +//! read end is gone. A dead server means a prompt error, never a hang. +//! +//! Writing to a FIFO with no read end open raises SIGPIPE, which kills a +//! process that left the signal at its default action. The client never +//! writes into that state: every FIFO it writes to has read access held for +//! the whole write, through the announcement descriptor or the request +//! descriptor itself. The server's writes rely on the Rust runtime instead, +//! which ignores SIGPIPE at startup, so a vanished client turns them into +//! plain errors. + +use std::{ + ffi::{OsStr, OsString}, + fs::{File, OpenOptions}, + io::{self, Read, Write}, + os::{fd::AsFd, unix::fs::OpenOptionsExt}, + pin::Pin, + task::{Context, Poll}, +}; + +use nix::{ + fcntl::{FcntlArg, OFlag, fcntl}, + poll::{PollFd, PollFlags, PollTimeout, poll}, + sys::stat::Mode, + unistd::mkfifo, +}; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}, + net::unix::pipe, +}; +use uuid::Uuid; +use vite_path::{AbsolutePath, AbsolutePathBuf}; + +const CONNECTION_ID_LEN: usize = 16; +const READY_BYTE: u8 = 1; +const CONNECT_FIFO_NAME: &str = "connect"; +const REQUEST_FIFO_SUFFIX: &str = ".request"; +const RESPONSE_FIFO_SUFFIX: &str = ".response"; + +pub struct Server { + _dir: tempfile::TempDir, + root: AbsolutePathBuf, + rendezvous: pipe::Receiver, + _rendezvous_keepalive: pipe::Sender, +} + +impl Server { + pub fn bind() -> io::Result { + let dir = tempfile::Builder::new().prefix("pipe_socket_").tempdir()?; + let root = AbsolutePath::new(dir.path()) + .expect("temp directories are absolute") + .to_absolute_path_buf(); + let rendezvous_path = connect_fifo(&root); + let mode = Mode::S_IRUSR | Mode::S_IWUSR; + mkfifo(rendezvous_path.as_path(), mode).map_err(io::Error::from)?; + + // Keep one sender open so an idle rendezvous FIFO does not report EOF + // between client announcements. A single read-write descriptor would + // do the same, but tokio only offers that on Linux. + let rendezvous = pipe::OpenOptions::new().open_receiver(&rendezvous_path)?; + let rendezvous_keepalive = pipe::OpenOptions::new().open_sender(&rendezvous_path)?; + + Ok(Self { _dir: dir, root, rendezvous, _rendezvous_keepalive: rendezvous_keepalive }) + } + + pub fn name(&self) -> &OsStr { + self.root.as_path().as_os_str() + } + + pub async fn accept(&mut self) -> io::Result { + loop { + let mut id = [0; CONNECTION_ID_LEN]; + self.rendezvous.read_exact(&mut id).await?; + let paths = connection_paths(&self.root, Uuid::from_bytes(id)); + + // A client can die between announcing itself and the handshake + // finishing. Both opens are nonblocking and fail in that case, + // and the failure belongs to that client alone: skip its + // announcement and wait for the next one. + let Ok(reader) = pipe::OpenOptions::new().open_receiver(&paths.request) else { + continue; + }; + let Ok(mut writer) = pipe::OpenOptions::new().open_sender(&paths.response) else { + continue; + }; + if writer.write_all(&[READY_BYTE]).await.is_err() { + continue; + } + + return Ok(ServerConnection { reader, writer }); + } + } +} + +pub struct ServerConnection { + reader: pipe::Receiver, + writer: pipe::Sender, +} + +impl AsyncRead for ServerConnection { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.reader).poll_read(cx, buf) + } +} + +impl AsyncWrite for ServerConnection { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.writer).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.writer).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.writer).poll_shutdown(cx) + } +} + +pub struct Client { + reader: File, + writer: File, +} + +impl Client { + pub fn connect(name: &OsStr) -> io::Result { + let root = AbsolutePath::new(name).ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "IPC server name is not absolute") + })?; + + // Open the rendezvous FIFO first, before creating anything. A blocking + // open would wait forever when the server is gone; a nonblocking write + // open fails with ENXIO instead, because no reader holds the other + // end. The descriptor is never written to; it exists for this check + // and for the death watch in the poll below. + let rendezvous = OpenOptions::new() + .write(true) + .custom_flags(OFlag::O_NONBLOCK.bits()) + .open(connect_fifo(root)) + .map_err(|error| { + if error.raw_os_error() == Some(nix::errno::Errno::ENXIO as i32) { + server_is_gone() + } else { + error + } + })?; + + let id = Uuid::new_v4(); + let paths = connection_paths(root, id); + let mode = Mode::S_IRUSR | Mode::S_IWUSR; + mkfifo(paths.request.as_path(), mode).map_err(io::Error::from)?; + mkfifo(paths.response.as_path(), mode).map_err(io::Error::from)?; + + // A read-write open never blocks, so the client does not wait for + // the server to open its read end. The client only writes to this + // descriptor, and the read access it holds means those writes can + // never raise SIGPIPE: after a server death they fill the FIFO's + // buffer, and the next response read reports end of file. + let writer = OpenOptions::new().read(true).write(true).open(&paths.request)?; + + // The connection's read end, nonblocking so this open does not wait + // for the server either. Read-write would be wrong here: the client + // must see end of file when the server dies, and that only happens + // when no write end is left open. The descriptor also serves the + // whole connection; reopening the FIFO after the ready byte could + // wait forever for a writer when the server wrote the byte and + // exited right after, while this descriptor reads the buffered byte + // with no writer around. + let mut reader = OpenOptions::new() + .read(true) + .custom_flags(OFlag::O_NONBLOCK.bits()) + .open(&paths.response)?; + + // The announcement goes through its own read-write descriptor: with + // read access held, the FIFO has a reader for the whole write, so a + // server that died after the liveness check cannot turn the write + // into a SIGPIPE kill. It drops right after the write; keeping a + // read end open would blind the death checks below, which report the + // server as alive while any reader exists. The fixed-size write is + // smaller than PIPE_BUF, so concurrent client IDs cannot interleave. + let mut announce = OpenOptions::new().read(true).write(true).open(connect_fifo(root))?; + loop { + match announce.write(id.as_bytes()) { + Ok(written) if written == id.as_bytes().len() => break, + Ok(_written) => { + return Err(io::Error::new( + io::ErrorKind::WriteZero, + "short IPC rendezvous write", + )); + } + Err(err) if err.kind() == io::ErrorKind::Interrupted => {} + Err(err) => return Err(err), + } + } + drop(announce); + + // Wait for the server's ready byte while watching for its death. A + // blocking read on the response FIFO could not see the death and + // would wait forever. + loop { + let mut fds = [ + PollFd::new(reader.as_fd(), PollFlags::POLLIN), + PollFd::new(rendezvous.as_fd(), PollFlags::empty()), + ]; + // The timeout drives the liveness probe below. Linux reports a + // dead server at once as POLLERR on the rendezvous write end; + // macOS poll does not always report FIFO events, so the probe is + // what catches it there. + match poll(&mut fds, PollTimeout::from(100u16)) { + Ok(_) => {} + // A signal interrupted the poll; retry it instead of failing + // the connect. + Err(nix::errno::Errno::EINTR) => continue, + Err(err) => return Err(err.into()), + } + + if fds[0].revents().is_some_and(|events| events.contains(PollFlags::POLLIN)) { + break; + } + if fds[1] + .revents() + .is_some_and(|events| events.intersects(PollFlags::POLLERR | PollFlags::POLLHUP)) + { + return Err(server_is_gone()); + } + // Reopening the rendezvous for writing fails once the server's + // read end is closed (ENXIO) or its directory is removed (ENOENT). + if let Err(error) = OpenOptions::new() + .write(true) + .custom_flags(OFlag::O_NONBLOCK.bits()) + .open(connect_fifo(root)) + { + return Err(match error.raw_os_error() { + Some(code) + if code == nix::errno::Errno::ENXIO as i32 + || code == nix::errno::Errno::ENOENT as i32 => + { + server_is_gone() + } + _ => error, + }); + } + } + drop(rendezvous); + + // The reader was opened nonblocking to make the open safe; from here + // on the connection wants plain blocking reads. + let flags = + OFlag::from_bits_retain(fcntl(&reader, FcntlArg::F_GETFL).map_err(io::Error::from)?); + fcntl(&reader, FcntlArg::F_SETFL(flags - OFlag::O_NONBLOCK)).map_err(io::Error::from)?; + + let mut ready = [0]; + reader.read_exact(&mut ready)?; + if ready[0] != READY_BYTE { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "invalid IPC rendezvous response", + )); + } + + Ok(Self { reader, writer }) + } +} + +fn server_is_gone() -> io::Error { + io::Error::new(io::ErrorKind::ConnectionRefused, "IPC server is gone") +} + +impl Read for Client { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.reader.read(buf) + } +} + +impl Write for Client { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.writer.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.writer.flush() + } +} + +struct ConnectionPaths { + request: AbsolutePathBuf, + response: AbsolutePathBuf, +} + +fn connect_fifo(root: &AbsolutePath) -> AbsolutePathBuf { + root.join(CONNECT_FIFO_NAME) +} + +fn connection_paths(root: &AbsolutePath, id: Uuid) -> ConnectionPaths { + let mut encoded = [0; 32]; + let encoded = id.simple().encode_lower(&mut encoded); + + let mut request_name = OsString::from(&*encoded); + request_name.push(REQUEST_FIFO_SUFFIX); + let mut response_name = OsString::from(&*encoded); + response_name.push(RESPONSE_FIFO_SUFFIX); + + ConnectionPaths { request: root.join(request_name), response: root.join(response_name) } +} diff --git a/crates/pipe_socket/src/windows.rs b/crates/pipe_socket/src/windows.rs new file mode 100644 index 000000000..d871a5a8e --- /dev/null +++ b/crates/pipe_socket/src/windows.rs @@ -0,0 +1,133 @@ +use std::{ + ffi::{OsStr, OsString}, + fs::File, + io::{self, Read, Write}, + os::windows::ffi::OsStrExt, + pin::Pin, + task::{Context, Poll}, +}; + +use tokio::{ + io::{AsyncRead, AsyncWrite, ReadBuf}, + net::windows::named_pipe::{NamedPipeServer, ServerOptions}, +}; +use winapi::um::namedpipeapi::WaitNamedPipeW; + +pub struct Server { + name: OsString, + /// The instance the next client will connect to. Replaced on every accept, + /// and created before the accepted instance is handed out, so concurrent + /// connect attempts never find the pipe without an instance. + pending: NamedPipeServer, +} + +impl Server { + pub fn bind() -> io::Result { + #[expect( + clippy::disallowed_macros, + reason = "the generated pipe name exceeds Str inline capacity" + )] + let name = OsString::from(format!(r"\\.\pipe\pipe_socket_{}", uuid::Uuid::new_v4())); + let pending = ServerOptions::new().first_pipe_instance(true).create(&name)?; + Ok(Self { name, pending }) + } + + pub fn name(&self) -> &OsStr { + &self.name + } + + pub async fn accept(&mut self) -> io::Result { + self.pending.connect().await?; + let next = ServerOptions::new().create(&self.name)?; + Ok(ServerConnection { inner: std::mem::replace(&mut self.pending, next) }) + } +} + +pub struct ServerConnection { + inner: NamedPipeServer, +} + +impl AsyncRead for ServerConnection { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl AsyncWrite for ServerConnection { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + +pub struct Client { + inner: File, +} + +impl Client { + /// Opens the named pipe as a client. + /// + /// Opening fails with `ERROR_PIPE_BUSY` when another client claimed the + /// server's only pending instance moments earlier, in the window between + /// the server accepting one connection and creating the next instance. + /// `WaitNamedPipeW` hands that wait to the kernel: it blocks until an + /// instance is available and fails when the pipe is gone. No polling and + /// no arbitrary timeouts. + pub fn connect(name: &OsStr) -> io::Result { + // ERROR_PIPE_BUSY, from WinError.h. `std::io::Error` has no typed + // constant for it. + const ERROR_PIPE_BUSY: i32 = 231; + // NMPWAIT_WAIT_FOREVER, from WinBase.h. winapi 0.3 does not define + // the NMPWAIT_* constants. + const NMPWAIT_WAIT_FOREVER: u32 = 0xFFFF_FFFF; + + let mut wide: Vec = name.encode_wide().collect(); + wide.push(0); + + loop { + match std::fs::OpenOptions::new().read(true).write(true).open(name) { + Ok(inner) => return Ok(Self { inner }), + Err(err) if err.raw_os_error() == Some(ERROR_PIPE_BUSY) => { + // SAFETY: `wide` is NUL-terminated and remains valid for + // the duration of the call. + let ok = unsafe { WaitNamedPipeW(wide.as_ptr(), NMPWAIT_WAIT_FOREVER) }; + if ok == 0 { + return Err(io::Error::last_os_error()); + } + } + Err(err) => return Err(err), + } + } + } +} + +impl Read for Client { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.inner.read(buf) + } +} + +impl Write for Client { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.inner.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} diff --git a/crates/pipe_socket/tests/integration.rs b/crates/pipe_socket/tests/integration.rs new file mode 100644 index 000000000..a81c4532f --- /dev/null +++ b/crates/pipe_socket/tests/integration.rs @@ -0,0 +1,175 @@ +use std::io::{Read as _, Write as _}; +#[cfg(unix)] +use std::sync::mpsc; + +use pipe_socket::{Client, Server}; +use tokio::{ + io::{AsyncReadExt as _, AsyncWriteExt as _}, + runtime::Builder, +}; + +#[test] +fn round_trip() { + let runtime = Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async { + let mut server = Server::bind().expect("bind server"); + let name = server.name().to_owned(); + + let client = tokio::task::spawn_blocking(move || { + let mut client = Client::connect(&name).expect("connect client"); + client.write_all(b"ping").expect("write request"); + client.flush().expect("flush request"); + + let mut response = [0; 4]; + client.read_exact(&mut response).expect("read response"); + assert_eq!(&response, b"pong"); + }); + + let mut connection = server.accept().await.expect("accept client"); + let mut request = [0; 4]; + connection.read_exact(&mut request).await.expect("read request"); + assert_eq!(&request, b"ping"); + connection.write_all(b"pong").await.expect("write response"); + connection.flush().await.expect("flush response"); + + client.await.expect("client task panicked"); + }); +} + +#[cfg(unix)] +#[test] +fn unix_uses_named_fifos() { + use std::os::unix::fs::FileTypeExt as _; + + let runtime = Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async { + let mut server = Server::bind().expect("bind server"); + let name = server.name().to_owned(); + let root = name.clone(); + let (connected_tx, connected_rx) = mpsc::channel(); + let (close_tx, close_rx) = mpsc::channel(); + + let client = tokio::task::spawn_blocking(move || { + let _client = Client::connect(&name).expect("connect client"); + connected_tx.send(()).expect("signal connected"); + close_rx.recv().expect("wait to close"); + }); + + let connection = server.accept().await.expect("accept client"); + tokio::task::spawn_blocking(move || connected_rx.recv().expect("wait for client")) + .await + .expect("wait task panicked"); + + let entries = std::fs::read_dir(root) + .expect("read IPC root") + .map(|entry| entry.expect("read IPC entry")) + .collect::>(); + assert_eq!(entries.len(), 3, "rendezvous + request/response FIFO pair"); + assert!( + entries.iter().all(|entry| entry.file_type().expect("read IPC entry type").is_fifo()), + "every Unix endpoint must be a named FIFO" + ); + + close_tx.send(()).expect("close client"); + client.await.expect("client task panicked"); + drop(connection); + }); +} + +/// A stale announcement from a client that died before the server got to it +/// must not fail the accept; the next client still connects. +#[cfg(unix)] +#[test] +fn accept_skips_announcement_from_dead_client() { + let runtime = Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async { + let mut server = Server::bind().expect("bind server"); + let name = server.name().to_owned(); + + // Leftovers of a client that died right after announcing itself: its + // FIFO pair exists on disk, its id sits in the rendezvous FIFO, and + // every descriptor it held is closed. + let id = [0xab; 16]; + let hex = "ab".repeat(16); + let mode = nix::sys::stat::Mode::S_IRUSR | nix::sys::stat::Mode::S_IWUSR; + nix::unistd::mkfifo(fifo_path(&name, &(hex.clone() + ".request")).as_os_str(), mode) + .expect("create request FIFO"); + nix::unistd::mkfifo(fifo_path(&name, &(hex + ".response")).as_os_str(), mode) + .expect("create response FIFO"); + std::fs::OpenOptions::new() + .write(true) + .open(fifo_path(&name, "connect")) + .expect("open rendezvous FIFO") + .write_all(&id) + .expect("announce dead client"); + + let client = tokio::task::spawn_blocking(move || { + let mut client = Client::connect(&name).expect("connect client"); + client.write_all(b"ping").expect("write request"); + + let mut response = [0; 4]; + client.read_exact(&mut response).expect("read response"); + assert_eq!(&response, b"pong"); + }); + + let accepted = tokio::time::timeout(std::time::Duration::from_secs(10), server.accept()); + let mut connection = accepted.await.expect("accept must not hang").expect("accept client"); + let mut request = [0; 4]; + connection.read_exact(&mut request).await.expect("read request"); + assert_eq!(&request, b"ping"); + connection.write_all(b"pong").await.expect("write response"); + + client.await.expect("client task panicked"); + }); +} + +#[cfg(unix)] +fn fifo_path(root: &std::ffi::OsStr, name: &str) -> std::ffi::OsString { + let mut path = root.to_os_string(); + path.push("/"); + path.push(name); + path +} + +/// The client must error out instead of hanging when the server disappears, +/// whether that happens before the connection attempt or in the middle of one. +#[test] +fn connect_fails_when_server_is_gone() { + let runtime = Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async { + // Server already gone: the name no longer resolves. + let server = Server::bind().expect("bind server"); + let name = server.name().to_owned(); + drop(server); + let stale = tokio::task::spawn_blocking(move || Client::connect(&name)); + let result = tokio::time::timeout(std::time::Duration::from_secs(10), stale) + .await + .expect("connect must not hang") + .expect("client task panicked"); + assert!(result.is_err()); + + // Server dies mid-connect, after the client reached it but before any + // accept: the client must notice instead of waiting for a ready byte + // that never comes. Only Unix has this wait state; a Windows client + // either opens the pipe or fails at once, which the case above covers. + #[cfg(unix)] + { + let server = Server::bind().expect("bind server"); + let name = server.name().to_owned(); + let root = name.clone(); + let midway = tokio::task::spawn_blocking(move || Client::connect(&name)); + // The client creates its request/response FIFO pair next to the + // rendezvous FIFO on its way in; three entries mean it reached + // the server. + while std::fs::read_dir(&root).map_or(0, Iterator::count) < 3 { + std::thread::yield_now(); + } + drop(server); + let result = tokio::time::timeout(std::time::Duration::from_secs(10), midway) + .await + .expect("connect must not hang") + .expect("client task panicked"); + assert!(result.is_err()); + } + }); +} diff --git a/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_anthropic_sandbox_runtime.md b/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_anthropic_sandbox_runtime.md index 5dfc32309..4746ea476 100644 --- a/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_anthropic_sandbox_runtime.md +++ b/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_anthropic_sandbox_runtime.md @@ -15,11 +15,9 @@ create the session temp that Claude Code supplies before sandboxed Bash commands ## `srt --settings claude-code-default-sandbox.json vt run inner` -**Exit code:** 1 - ``` $ vtt print-file input.txt -✗ Failed to set up task communication: Operation not permitted (os error 1) +tracked input ``` ## `vtt replace-file-content input.txt tracked modified` @@ -29,9 +27,7 @@ $ vtt print-file input.txt ## `srt --settings claude-code-default-sandbox.json vt run inner` -**Exit code:** 1 - ``` -$ vtt print-file input.txt -✗ Failed to set up task communication: Operation not permitted (os error 1) +$ vtt print-file input.txt ○ cache miss: 'input.txt' modified, executing +modified input ``` diff --git a/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_codex_sandbox.md b/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_codex_sandbox.md index 05426b33b..abafdc349 100644 --- a/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_codex_sandbox.md +++ b/crates/vite_task_bin/tests/e2e_snapshots/fixtures/sandboxed_fspy/snapshots/fspy_under_codex_sandbox.md @@ -6,11 +6,9 @@ The nested `vt` enables fspy for automatic input inference; changing the file re ## `codex sandbox -P :workspace vt run inner` -**Exit code:** 1 - ``` $ vtt print-file input.txt -✗ Failed to set up task communication: Operation not permitted (os error 1) +tracked input ``` ## `vtt replace-file-content input.txt tracked modified` @@ -20,9 +18,7 @@ $ vtt print-file input.txt ## `codex sandbox -P :workspace vt run inner` -**Exit code:** 1 - ``` -$ vtt print-file input.txt -✗ Failed to set up task communication: Operation not permitted (os error 1) +$ vtt print-file input.txt ○ cache miss: 'input.txt' modified, executing +modified input ``` diff --git a/crates/vite_task_client/Cargo.toml b/crates/vite_task_client/Cargo.toml index b926bc01f..171a52cac 100644 --- a/crates/vite_task_client/Cargo.toml +++ b/crates/vite_task_client/Cargo.toml @@ -9,13 +9,11 @@ rust-version.workspace = true [dependencies] native_str = { workspace = true } rustc-hash = { workspace = true } +pipe_socket = { workspace = true } vite_path = { workspace = true } vite_task_ipc_shared = { workspace = true } wincode = { workspace = true, features = ["derive"] } -[target.'cfg(windows)'.dependencies] -winapi = { workspace = true, features = ["namedpipeapi"] } - [lints] workspace = true diff --git a/crates/vite_task_client/src/lib.rs b/crates/vite_task_client/src/lib.rs index 715de5c08..bdde8de66 100644 --- a/crates/vite_task_client/src/lib.rs +++ b/crates/vite_task_client/src/lib.rs @@ -6,6 +6,7 @@ use std::{ }; use native_str::NativeStr; +use pipe_socket::Client as Stream; use rustc_hash::FxHashMap; use vite_path::{self, AbsolutePath}; use vite_task_ipc_shared::{ @@ -13,11 +14,6 @@ use vite_task_ipc_shared::{ }; use wincode::{SchemaRead, config::DefaultConfig}; -#[cfg(unix)] -type Stream = std::os::unix::net::UnixStream; -#[cfg(windows)] -type Stream = std::fs::File; - pub struct Client { stream: RefCell, scratch: RefCell>, @@ -44,7 +40,7 @@ impl Client { ) -> io::Result> { for (name, value) in envs { if name.as_ref() == IPC_ENV_NAME { - let stream = connect(value.as_ref())?; + let stream = Stream::connect(value.as_ref())?; return Ok(Some(Self::from_stream(stream))); } } @@ -188,55 +184,3 @@ fn resolve_path(path: &OsStr) -> io::Result> { absolute.push(path); Ok(Box::::from(absolute.as_absolute_path().as_path().as_os_str())) } - -#[cfg(unix)] -fn connect(name: &OsStr) -> io::Result { - std::os::unix::net::UnixStream::connect(name) -} - -/// Open a Windows named pipe as a client. -/// -/// `OpenOptions::open` on a named-pipe path fails with `ERROR_PIPE_BUSY` when -/// the server's only pending instance has just been claimed by another client -/// — the brief window between the server accepting one connection and creating -/// the next instance. On `ERROR_PIPE_BUSY` we hand off to the kernel via -/// `WaitNamedPipeW`, which blocks until an instance becomes available (or -/// fails if the named pipe is gone). No polling and no arbitrary timeouts. -/// -/// This matches what the `interprocess` crate does internally. -#[cfg(windows)] -fn connect(name: &OsStr) -> io::Result { - use std::{fs::OpenOptions, os::windows::ffi::OsStrExt}; - - use winapi::um::namedpipeapi::WaitNamedPipeW; - - // ERROR_PIPE_BUSY — see WinError.h. `std::io::Error` does not expose a - // typed constant for this, so the raw OS code is the cleanest test. - const ERROR_PIPE_BUSY: i32 = 231; - // NMPWAIT_WAIT_FOREVER — see WinBase.h. winapi 0.3 doesn't define the - // NMPWAIT_* constants yet (only the comment placeholder). - const NMPWAIT_WAIT_FOREVER: u32 = 0xFFFF_FFFF; - - // `WaitNamedPipeW` needs a NUL-terminated UTF-16 path. - let mut wide: Vec = name.encode_wide().collect(); - wide.push(0); - - loop { - match OpenOptions::new().read(true).write(true).open(name) { - Ok(file) => return Ok(file), - Err(err) if err.raw_os_error() == Some(ERROR_PIPE_BUSY) => { - // SAFETY: `wide` is NUL-terminated; pointer stays valid for - // the call's duration. `NMPWAIT_WAIT_FOREVER` makes this a - // bounded kernel wait (server's pipe wait-timeout is the - // upper bound on each retry; default ~50ms, then we loop). - let ok = unsafe { WaitNamedPipeW(wide.as_ptr(), NMPWAIT_WAIT_FOREVER) }; - if ok == 0 { - return Err(io::Error::last_os_error()); - } - // Loop and re-open — another client may have raced us to the - // newly-available instance. - } - Err(err) => return Err(err), - } - } -} diff --git a/crates/vite_task_server/Cargo.toml b/crates/vite_task_server/Cargo.toml index 94f1bad8d..fd06856da 100644 --- a/crates/vite_task_server/Cargo.toml +++ b/crates/vite_task_server/Cargo.toml @@ -10,19 +10,16 @@ rust-version.workspace = true futures = { workspace = true } native_str = { workspace = true } rustc-hash = { workspace = true } -tempfile = { workspace = true } thiserror = { workspace = true } -tokio = { workspace = true, features = ["io-util", "net", "rt", "macros"] } +tokio = { workspace = true, features = ["io-util", "rt", "macros"] } tokio-util = { workspace = true } tracing = { workspace = true } vite_glob = { workspace = true } +pipe_socket = { workspace = true } vite_path = { workspace = true } vite_task_ipc_shared = { workspace = true } wincode = { workspace = true, features = ["derive"] } -[target.'cfg(windows)'.dependencies] -uuid = { workspace = true, features = ["v4"] } - [dev-dependencies] tokio = { workspace = true, features = ["io-util", "net", "rt", "macros", "time"] } vite_task_client = { workspace = true } diff --git a/crates/vite_task_server/src/lib.rs b/crates/vite_task_server/src/lib.rs index b33ac877a..8f0d9835b 100644 --- a/crates/vite_task_server/src/lib.rs +++ b/crates/vite_task_server/src/lib.rs @@ -7,6 +7,7 @@ use std::{ use futures::{FutureExt, StreamExt, future::LocalBoxFuture, stream::FuturesUnordered}; use native_str::NativeStr; +use pipe_socket::{Server as TransportServer, ServerConnection as Stream}; use rustc_hash::{FxHashMap, FxHashSet}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_util::sync::CancellationToken; @@ -265,13 +266,13 @@ impl StopAccepting { /// /// # Errors /// -/// Returns an error if creating the listener fails (on Unix, this includes -/// creating the temp socket path). +/// Returns an error if creating the transport server fails. pub fn serve<'h, H: Handler + 'h>( handler: H, ) -> io::Result<(impl Iterator, ServerHandle<'h, H>)> { let stop_token = CancellationToken::new(); - let (name, bound) = bind_listener()?; + let server = TransportServer::bind()?; + let name = server.name().to_owned(); let run_stop = stop_token.clone(); let driver = async move { @@ -282,7 +283,7 @@ pub fn serve<'h, H: Handler + 'h>( // single-threaded runtime and handler methods are synchronous (no // awaits, so no borrow spans a yield point). let handler = RefCell::new(handler); - let first_err = run(bound, &handler, run_stop).await; + let first_err = run(server, &handler, run_stop).await; first_err.map_or_else(|| Ok(handler.into_inner()), Err) } .boxed_local(); @@ -293,83 +294,8 @@ pub fn serve<'h, H: Handler + 'h>( )) } -#[cfg(unix)] -type Stream = tokio::net::UnixStream; -#[cfg(windows)] -type Stream = tokio::net::windows::named_pipe::NamedPipeServer; - -/// The bound listener for the IPC server. -/// -/// Unix: a Tokio [`UnixListener`](tokio::net::UnixListener) bound inside a -/// [`NamedTempFile`](tempfile::NamedTempFile) so its socket file is unlinked -/// on `Drop`. Windows: a single named-pipe instance that is created up front -/// and replaced on each `accept` (a new pipe instance must be created before -/// the previous one is handed to the client, otherwise concurrent connect -/// attempts race for it). -#[cfg(unix)] -struct Bound { - file: tempfile::NamedTempFile, -} - -#[cfg(windows)] -struct Bound { - pipe_name: OsString, - pending: tokio::net::windows::named_pipe::NamedPipeServer, -} - -#[cfg(unix)] -fn bind_listener() -> io::Result<(OsString, Bound)> { - // `make` lets us bind the socket directly to the path tempfile picks; the - // closure is responsible for creating the file (`UnixListener::bind` does). - // The `NamedTempFile` wrapper unlinks the socket path on `Drop`. - let file = tempfile::Builder::new() - .prefix("vite_task_ipc_") - .make(|path| tokio::net::UnixListener::bind(path))?; - let name = file.path().as_os_str().to_owned(); - Ok((name, Bound { file })) -} - -#[cfg(windows)] -fn bind_listener() -> io::Result<(OsString, Bound)> { - use tokio::net::windows::named_pipe::ServerOptions; - - #[expect( - clippy::disallowed_macros, - reason = "pipe name always exceeds Str inline capacity; format! is the simplest construction" - )] - let pipe_name = OsString::from(format!(r"\\.\pipe\vite_task_ipc_{}", uuid::Uuid::new_v4())); - let pending = ServerOptions::new().first_pipe_instance(true).create(&pipe_name)?; - Ok((pipe_name.clone(), Bound { pipe_name, pending })) -} - -impl Bound { - #[cfg(unix)] - #[expect( - clippy::needless_pass_by_ref_mut, - reason = "Windows variant requires &mut self to swap pending instance; keep the signature uniform across cfgs so `run` can call it identically." - )] - async fn accept(&mut self) -> io::Result { - let (stream, _addr) = self.file.as_file().accept().await?; - Ok(stream) - } - - #[cfg(windows)] - async fn accept(&mut self) -> io::Result { - use tokio::net::windows::named_pipe::ServerOptions; - - // Wait for the next client to connect to the currently-pending - // instance, then immediately create a fresh instance to listen for the - // connection after that. Creating the next instance before yielding the - // accepted one ensures no client gets `ERROR_PIPE_BUSY` during the - // handoff. - self.pending.connect().await?; - let next = ServerOptions::new().create(&self.pipe_name)?; - Ok(std::mem::replace(&mut self.pending, next)) - } -} - async fn run( - mut bound: Bound, + mut server: TransportServer, handler: &RefCell, shutdown: CancellationToken, ) -> Option { @@ -380,7 +306,7 @@ async fn run( loop { tokio::select! { () = shutdown.cancelled() => break, - accept_result = bound.accept() => { + accept_result = server.accept() => { match accept_result { Ok(stream) => { clients.push(handle_client(stream, handler).boxed_local()); @@ -401,9 +327,8 @@ async fn run( } } - // Stop accepting: drop the listener (and on Unix unlink the socket file). - // Existing client streams continue to work. - drop(bound); + // Stop accepting. Existing client streams continue to work. + drop(server); // Drain phase: wait for all in-flight per-client tasks to finish. while let Some(result) = clients.next().await { diff --git a/crates/vite_task_server/tests/integration.rs b/crates/vite_task_server/tests/integration.rs index e22581812..7b84d3ef8 100644 --- a/crates/vite_task_server/tests/integration.rs +++ b/crates/vite_task_server/tests/integration.rs @@ -6,11 +6,7 @@ use std::{ }; use native_str::NativeStr; - -#[cfg(unix)] -type RawStream = std::os::unix::net::UnixStream; -#[cfg(windows)] -type RawStream = std::fs::File; +use pipe_socket::Client as RawStream; use rustc_hash::FxHashMap; use tokio::runtime::Builder; use vite_task_client::{Client, GetEnvsQuery}; @@ -64,14 +60,8 @@ fn flush(client: &Client) { let _ = client.get_env(OsStr::new("__VP_TEST_FLUSH__"), false).unwrap(); } -#[cfg(unix)] -fn connect_raw(name: &OsStr) -> RawStream { - std::os::unix::net::UnixStream::connect(name).expect("connect raw") -} - -#[cfg(windows)] fn connect_raw(name: &OsStr) -> RawStream { - std::fs::OpenOptions::new().read(true).write(true).open(name).expect("connect raw") + RawStream::connect(name).expect("connect raw") } fn send_frame(stream: &mut RawStream, request: &Request<'_>) {