diff --git a/CHANGELOG.md b/CHANGELOG.md index 91b6764e..a58ad9e8 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 11097429..ab1b67e8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4285,8 +4285,8 @@ dependencies = [ "native_str", "rustc-hash", "vite_path", + "vite_task_ipc", "vite_task_ipc_shared", - "winapi", "wincode", ] @@ -4324,6 +4324,18 @@ dependencies = [ "wincode", ] +[[package]] +name = "vite_task_ipc" +version = "0.0.0" +dependencies = [ + "nix 0.31.2", + "tempfile", + "tokio", + "uuid", + "vite_path", + "winapi", +] + [[package]] name = "vite_task_ipc_shared" version = "0.0.0" @@ -4378,15 +4390,14 @@ dependencies = [ "futures", "native_str", "rustc-hash", - "tempfile", "thiserror 2.0.18", "tokio", "tokio-util", "tracing", - "uuid", "vite_glob", "vite_path", "vite_task_client", + "vite_task_ipc", "vite_task_ipc_shared", "wincode", ] diff --git a/Cargo.toml b/Cargo.toml index 1e12e8b9..a2fb465a 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" } +vite_task_ipc = { path = "crates/vite_task_ipc" } vite_path = { path = "crates/vite_path" } vite_powershell = { path = "crates/vite_powershell" } vite_select = { path = "crates/vite_select" } 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 5dfc3230..4746ea47 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 05426b33..abafdc34 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 b926bc01..dc9173e5 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 } +vite_task_ipc = { 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 715de5c0..dae793ef 100644 --- a/crates/vite_task_client/src/lib.rs +++ b/crates/vite_task_client/src/lib.rs @@ -8,16 +8,12 @@ use std::{ use native_str::NativeStr; use rustc_hash::FxHashMap; use vite_path::{self, AbsolutePath}; +use vite_task_ipc::Client as Stream; use vite_task_ipc_shared::{ EnvQuery as IpcEnvQuery, GetEnvResponse, GetEnvsResponse, IPC_ENV_NAME, Request, }; 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_ipc/Cargo.toml b/crates/vite_task_ipc/Cargo.toml new file mode 100644 index 00000000..289cd807 --- /dev/null +++ b/crates/vite_task_ipc/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "vite_task_ipc" +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 } +tokio = { workspace = true, features = ["fs"] } +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"] } + +[lints] +workspace = true + +[lib] +doctest = false +test = false diff --git a/crates/vite_task_ipc/README.md b/crates/vite_task_ipc/README.md new file mode 100644 index 00000000..5833920e --- /dev/null +++ b/crates/vite_task_ipc/README.md @@ -0,0 +1,15 @@ +# `vite_task_ipc` + +Name-based byte transport between `vp run` and the processes it spawns. + +The server binds under an opaque name and passes that name to child processes +through an environment variable. Clients connect synchronously; the server +accepts asynchronously with Tokio. + +Unix uses named FIFOs instead of Unix domain sockets, because coding-agent +sandboxes such as the default Codex CLI and Claude Code sandboxes block the +sockets and allow the FIFOs. Windows uses named pipes. Both transports live +behind the same four-item API: `Server::bind`, `Server::name`, +`Server::accept`, and `Client::connect`. + +A client whose server is gone gets an error, not a hang, on both platforms. diff --git a/crates/vite_task_ipc/src/lib.rs b/crates/vite_task_ipc/src/lib.rs new file mode 100644 index 00000000..48ef8f40 --- /dev/null +++ b/crates/vite_task_ipc/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!("vite_task_ipc 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/vite_task_ipc/src/unix.rs b/crates/vite_task_ipc/src/unix.rs new file mode 100644 index 00000000..cf151e7d --- /dev/null +++ b/crates/vite_task_ipc/src/unix.rs @@ -0,0 +1,344 @@ +//! 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. +//! 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, which needs care in two +//! places. The client holds short-lived nonblocking read ends of its own +//! FIFOs, so both processes can open their write ends without a deadline on +//! who goes first. And the client never waits on the server blindly: the +//! rendezvous open is nonblocking, and the wait for the ready byte also +//! 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. + +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}, + sys::stat::Mode, + unistd::mkfifo, +}; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, 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("vite_task_ipc_").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. + 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 { + let mut id = [0; CONNECTION_ID_LEN]; + self.rendezvous.read_exact(&mut id).await?; + let paths = connection_paths(&self.root, ConnectionId::from_bytes(id)); + + // Both opens are nonblocking, so a client that died after announcing + // itself produces an error here instead of a stuck accept. The caller + // logs it and keeps accepting. + let reader = open_fifo_receiver(&paths.request)?; + let mut writer = open_fifo_sender(&paths.response)?; + writer.write_all(&[READY_BYTE])?; + writer.flush()?; + + Ok(ServerConnection { + reader: tokio::fs::File::from_std(reader), + writer: tokio::fs::File::from_std(writer), + }) + } +} + +pub struct ServerConnection { + reader: tokio::fs::File, + writer: tokio::fs::File, +} + +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. + let mut 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 + } + })?; + set_blocking(&rendezvous)?; + + let id = ConnectionId::random(); + 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)?; + + // The temporary readers let both writers open without blocking. They + // stay alive until the ready byte confirms that the server has opened + // its ends of both FIFOs. + let request_bootstrap = OpenOptions::new() + .read(true) + .custom_flags(OFlag::O_NONBLOCK.bits()) + .open(&paths.request)?; + let writer = OpenOptions::new().write(true).open(&paths.request)?; + let response_bootstrap = OpenOptions::new() + .read(true) + .custom_flags(OFlag::O_NONBLOCK.bits()) + .open(&paths.response)?; + + write_connection_id(&mut rendezvous, id)?; + + // 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. + wait_for_ready(root, &rendezvous, &response_bootstrap)?; + + // The ready byte is in the response FIFO and the server has both of + // its ends open, so this open returns at once and the byte is read + // from the connection's own descriptor. + let mut reader = OpenOptions::new().read(true).open(&paths.response)?; + drop(response_bootstrap); + drop(rendezvous); + + 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", + )); + } + drop(request_bootstrap); + + Ok(Self { reader, writer }) + } +} + +/// Blocks until the response FIFO has the ready byte or the server dies. +fn wait_for_ready(root: &AbsolutePath, rendezvous: &File, response: &File) -> io::Result<()> { + use nix::poll::{PollFd, PollFlags, PollTimeout, poll}; + + loop { + let mut fds = [ + PollFd::new(response.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. + poll(&mut fds, PollTimeout::from(100u16)).map_err(io::Error::from)?; + + if fds[0].revents().is_some_and(|events| events.contains(PollFlags::POLLIN)) { + return Ok(()); + } + 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, + }); + } + } +} + +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() + } +} + +#[derive(Clone, Copy)] +struct ConnectionId(Uuid); + +impl ConnectionId { + fn random() -> Self { + Self(Uuid::new_v4()) + } + + const fn from_bytes(bytes: [u8; CONNECTION_ID_LEN]) -> Self { + Self(Uuid::from_bytes(bytes)) + } + + const fn as_bytes(&self) -> &[u8; CONNECTION_ID_LEN] { + self.0.as_bytes() + } +} + +struct ConnectionPaths { + request: AbsolutePathBuf, + response: AbsolutePathBuf, +} + +fn connect_fifo(root: &AbsolutePath) -> AbsolutePathBuf { + root.join(CONNECT_FIFO_NAME) +} + +fn connection_paths(root: &AbsolutePath, id: ConnectionId) -> ConnectionPaths { + let mut encoded = [0; 32]; + let encoded = id.0.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) } +} + +fn write_connection_id(rendezvous: &mut File, id: ConnectionId) -> io::Result<()> { + // This fixed-size write is smaller than PIPE_BUF, so concurrent client IDs + // cannot interleave. + loop { + match rendezvous.write(id.as_bytes()) { + Ok(written) if written == id.as_bytes().len() => return Ok(()), + 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), + } + } +} + +fn open_fifo_receiver(path: &AbsolutePath) -> io::Result { + let file = OpenOptions::new().read(true).custom_flags(OFlag::O_NONBLOCK.bits()).open(path)?; + set_blocking(&file)?; + Ok(file) +} + +fn open_fifo_sender(path: &AbsolutePath) -> io::Result { + let file = OpenOptions::new().write(true).custom_flags(OFlag::O_NONBLOCK.bits()).open(path)?; + set_blocking(&file)?; + Ok(file) +} + +fn set_blocking(file: &File) -> io::Result<()> { + let flags = OFlag::from_bits_retain(fcntl(file, FcntlArg::F_GETFL).map_err(io::Error::from)?); + fcntl(file, FcntlArg::F_SETFL(flags - OFlag::O_NONBLOCK)).map_err(io::Error::from)?; + Ok(()) +} diff --git a/crates/vite_task_ipc/src/windows.rs b/crates/vite_task_ipc/src/windows.rs new file mode 100644 index 00000000..ba1029e4 --- /dev/null +++ b/crates/vite_task_ipc/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\vite_task_ipc_{}", 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/vite_task_ipc/tests/integration.rs b/crates/vite_task_ipc/tests/integration.rs new file mode 100644 index 00000000..2df95418 --- /dev/null +++ b/crates/vite_task_ipc/tests/integration.rs @@ -0,0 +1,116 @@ +use std::io::{Read as _, Write as _}; +#[cfg(unix)] +use std::sync::mpsc; + +use tokio::{ + io::{AsyncReadExt as _, AsyncWriteExt as _}, + runtime::Builder, +}; +use vite_task_ipc::{Client, Server}; + +#[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); + }); +} + +/// 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. + 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_server/Cargo.toml b/crates/vite_task_server/Cargo.toml index 94f1bad8..328cb266 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 } +vite_task_ipc = { 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 b33ac877..e6d23917 100644 --- a/crates/vite_task_server/src/lib.rs +++ b/crates/vite_task_server/src/lib.rs @@ -11,6 +11,7 @@ use rustc_hash::{FxHashMap, FxHashSet}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_util::sync::CancellationToken; use vite_path::AbsolutePath; +use vite_task_ipc::{Server as TransportServer, ServerConnection as Stream}; use vite_task_ipc_shared::{ EnvQuery as IpcEnvQuery, GetEnvResponse, GetEnvsResponse, IPC_ENV_NAME, Request, }; @@ -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 e2258181..fd3cdbf7 100644 --- a/crates/vite_task_server/tests/integration.rs +++ b/crates/vite_task_server/tests/integration.rs @@ -6,14 +6,10 @@ use std::{ }; use native_str::NativeStr; - -#[cfg(unix)] -type RawStream = std::os::unix::net::UnixStream; -#[cfg(windows)] -type RawStream = std::fs::File; use rustc_hash::FxHashMap; use tokio::runtime::Builder; use vite_task_client::{Client, GetEnvsQuery}; +use vite_task_ipc::Client as RawStream; use vite_task_ipc_shared::{GetEnvResponse, Request}; use vite_task_server::{EnvQuery, Error, Recorder, Reports, ServerHandle, serve}; @@ -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<'_>) {