diff --git a/CHANGELOG.md b/CHANGELOG.md index a76e814d..317b3eb5 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; Unix now uses named FIFOs ([#562](https://github.com/voidzero-dev/vite-task/issues/562)). - **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)). - **Fixed** npm workspace patterns beginning with `./` now discover matching packages correctly ([vite-plus#2201](https://github.com/voidzero-dev/vite-plus/issues/2201), [#547](https://github.com/voidzero-dev/vite-task/pull/547)). - **Fixed** Failures while forwarding output from a started task process no longer incorrectly say the process failed to spawn ([#506](https://github.com/voidzero-dev/vite-task/issues/506)). diff --git a/Cargo.lock b/Cargo.lock index fd72b2ce..3581b181 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4125,6 +4125,18 @@ dependencies = [ "serde_json", ] +[[package]] +name = "vite_ipc" +version = "0.0.0" +dependencies = [ + "nix 0.31.2", + "tempfile", + "tokio", + "uuid", + "vite_path", + "winapi", +] + [[package]] name = "vite_path" version = "0.1.0" @@ -4274,9 +4286,9 @@ version = "0.0.0" dependencies = [ "native_str", "rustc-hash", + "vite_ipc", "vite_path", "vite_task_ipc_shared", - "winapi", "wincode", ] @@ -4368,13 +4380,12 @@ dependencies = [ "futures", "native_str", "rustc-hash", - "tempfile", "thiserror 2.0.18", "tokio", "tokio-util", "tracing", - "uuid", "vite_glob", + "vite_ipc", "vite_path", "vite_task_client", "vite_task_ipc_shared", diff --git a/Cargo.toml b/Cargo.toml index 80454123..20c4bfbd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -148,6 +148,7 @@ uuid = "1.18.1" vec1 = "1.12.1" vite_glob = { path = "crates/vite_glob" } vite_graph_ser = { path = "crates/vite_graph_ser" } +vite_ipc = { path = "crates/vite_ipc" } vite_path = { path = "crates/vite_path" } vite_powershell = { path = "crates/vite_powershell" } vite_select = { path = "crates/vite_select" } diff --git a/crates/vite_ipc/Cargo.toml b/crates/vite_ipc/Cargo.toml new file mode 100644 index 00000000..d2aa2933 --- /dev/null +++ b/crates/vite_ipc/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "vite_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"] } +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"] } + +[lints] +workspace = true + +[lib] +doctest = false +test = false diff --git a/crates/vite_ipc/README.md b/crates/vite_ipc/README.md new file mode 100644 index 00000000..275d3af6 --- /dev/null +++ b/crates/vite_ipc/README.md @@ -0,0 +1,8 @@ +# `vite_ipc` + +Name-based cross-platform byte transport for communication between a server +and its child processes. + +The server exposes an opaque name that can be passed through an environment +variable or process argument. Clients connect synchronously, while the server +accepts connections asynchronously with Tokio. diff --git a/crates/vite_ipc/src/lib.rs b/crates/vite_ipc/src/lib.rs new file mode 100644 index 00000000..d6f5f6c3 --- /dev/null +++ b/crates/vite_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_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_ipc/src/unix.rs b/crates/vite_ipc/src/unix.rs new file mode 100644 index 00000000..a1ca00bb --- /dev/null +++ b/crates/vite_ipc/src/unix.rs @@ -0,0 +1,242 @@ +use std::{ + ffi::{OsStr, OsString}, + fs::{File, OpenOptions}, + io::{self, Read, Write}, + os::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_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)); + + 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") + })?; + 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)?; + + let mut rendezvous = OpenOptions::new().write(true).open(connect_fifo(root))?; + write_connection_id(&mut rendezvous, id)?; + drop(rendezvous); + + let mut reader = OpenOptions::new().read(true).open(&paths.response)?; + drop(response_bootstrap); + + 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 }) + } +} + +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_ipc/src/windows.rs b/crates/vite_ipc/src/windows.rs new file mode 100644 index 00000000..ffa099b5 --- /dev/null +++ b/crates/vite_ipc/src/windows.rs @@ -0,0 +1,121 @@ +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, + 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_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 { + pub fn connect(name: &OsStr) -> io::Result { + // ERROR_PIPE_BUSY — see WinError.h. `std::io::Error` does not expose a + // typed constant for it. + const ERROR_PIPE_BUSY: i32 = 231; + // NMPWAIT_WAIT_FOREVER — winapi 0.3 does not define NMPWAIT_*. + 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_ipc/tests/integration.rs b/crates/vite_ipc/tests/integration.rs new file mode 100644 index 00000000..d2373459 --- /dev/null +++ b/crates/vite_ipc/tests/integration.rs @@ -0,0 +1,77 @@ +use std::io::{Read as _, Write as _}; +#[cfg(unix)] +use std::sync::mpsc; + +use tokio::{ + io::{AsyncReadExt as _, AsyncWriteExt as _}, + runtime::Builder, +}; +use vite_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); + }); +} 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..a18928b3 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 @@ -10,7 +10,7 @@ The nested `vt` enables fspy for automatic input inference; changing the file re ``` $ vtt print-file input.txt -✗ Failed to set up task communication: Operation not permitted (os error 1) +✗ Failed to spawn process: failed to create IPC channel: Operation not permitted (os error 1) ``` ## `vtt replace-file-content input.txt tracked modified` @@ -24,5 +24,5 @@ $ vtt print-file input.txt ``` $ vtt print-file input.txt -✗ Failed to set up task communication: Operation not permitted (os error 1) +✗ Failed to spawn process: failed to create IPC channel: Operation not permitted (os error 1) ``` diff --git a/crates/vite_task_client/Cargo.toml b/crates/vite_task_client/Cargo.toml index b926bc01..9354d372 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_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..f637f915 100644 --- a/crates/vite_task_client/src/lib.rs +++ b/crates/vite_task_client/src/lib.rs @@ -7,17 +7,13 @@ use std::{ use native_str::NativeStr; use rustc_hash::FxHashMap; +use vite_ipc::Client as Stream; use vite_path::{self, AbsolutePath}; 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_server/Cargo.toml b/crates/vite_task_server/Cargo.toml index 94f1bad8..84ab72ae 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_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..6faf23a9 100644 --- a/crates/vite_task_server/src/lib.rs +++ b/crates/vite_task_server/src/lib.rs @@ -10,6 +10,7 @@ use native_str::NativeStr; use rustc_hash::{FxHashMap, FxHashSet}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_util::sync::CancellationToken; +use vite_ipc::{Server as TransportServer, ServerConnection as Stream}; use vite_path::AbsolutePath; 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..9bd3c8e3 100644 --- a/crates/vite_task_server/tests/integration.rs +++ b/crates/vite_task_server/tests/integration.rs @@ -6,13 +6,9 @@ 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_ipc::Client as RawStream; use vite_task_client::{Client, GetEnvsQuery}; 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<'_>) {