diff --git a/Cargo.lock b/Cargo.lock index d39168c8..d3c3461a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2228,6 +2228,15 @@ dependencies = [ "libc", ] +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "miniz_oxide" version = "0.8.8" @@ -2840,8 +2849,8 @@ dependencies = [ "sqlite", "tempfile", "thiserror 2.0.17", + "uds_windows", "unicode-normalization", - "winpipe", ] [[package]] @@ -3071,7 +3080,7 @@ dependencies = [ "tempfile", "test-log", "thiserror 2.0.17", - "winpipe", + "uds_windows", ] [[package]] @@ -4491,6 +4500,17 @@ version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825" +[[package]] +name = "uds_windows" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89daebc3e6fd160ac4aa9fc8b3bf71e1f74fbf92367ae71fb83a037e8bf164b9" +dependencies = [ + "memoffset", + "tempfile", + "winapi", +] + [[package]] name = "unarray" version = "0.1.4" diff --git a/Cargo.toml b/Cargo.toml index 4e8f380b..b5604abb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -69,6 +69,7 @@ snapbox = "0.4.3" sqlite = "0.32.0" tempfile = "3.3.0" thiserror = { version = "2", default-features = false } +uds_windows = "1.1.0" winpipe = "0.1.1" winsplit = "0.1.0" zeroize = "1.5.7" diff --git a/crates/radicle-node/Cargo.toml b/crates/radicle-node/Cargo.toml index b49bb904..439dac91 100644 --- a/crates/radicle-node/Cargo.toml +++ b/crates/radicle-node/Cargo.toml @@ -47,8 +47,8 @@ thiserror = { workspace = true, default-features = true } radicle-systemd = { workspace = true, optional = true } [target.'cfg(windows)'.dependencies] -winpipe = { workspace = true } radicle-windows = { workspace = true } +uds_windows = { workspace = true } [dev-dependencies] mio = { version = "1", features = ["os-ext"] } diff --git a/crates/radicle-node/src/control.rs b/crates/radicle-node/src/control.rs index 434ff6c9..9ed568dd 100644 --- a/crates/radicle-node/src/control.rs +++ b/crates/radicle-node/src/control.rs @@ -6,9 +6,9 @@ use std::path::PathBuf; use std::{io, net, time}; #[cfg(unix)] -use std::os::unix::net::{UnixListener as Listener, UnixStream as Stream}; +use std::os::unix::net::{UnixListener, UnixStream}; #[cfg(windows)] -use winpipe::{WinListener as Listener, WinStream as Stream}; +use uds_windows::{UnixListener, UnixStream}; use radicle::node::Handle; use serde_json as json; @@ -33,7 +33,7 @@ pub enum Error { } /// Listen for commands on the control socket, and process them. -pub fn listen(listener: Listener, handle: H) -> Result<(), Error> +pub fn listen(listener: UnixListener, handle: H) -> Result<(), Error> where H: Handle + 'static, H::Sessions: serde::Serialize, @@ -45,11 +45,11 @@ where for incoming in listener.incoming() { match incoming { - Ok(stream) => { + Ok(mut stream) => { let handle = handle.clone(); thread::spawn(&nid, "control", move || { - if let Err((e, mut stream)) = command(stream, handle) { + if let Err(e) = command(&stream, handle) { log::debug!(target: "control", "Command returned error: {e}"); CommandResult::error(e).to_writer(&mut stream).ok(); @@ -77,63 +77,16 @@ enum CommandError { Io(#[from] io::Error), } -#[cfg(unix)] -fn command(stream: Stream, handle: H) -> Result<(), (CommandError, Stream)> +fn command(stream: &UnixStream, mut handle: H) -> Result<(), CommandError> where H: Handle + 'static, H::Sessions: serde::Serialize, CommandResult: From, E: serde::Serialize, { - let reader = BufReader::new(&stream); - let writer = LineWriter::new(&stream); + let mut reader = BufReader::new(stream); + let mut writer = LineWriter::new(stream); - command_internal(reader, writer, handle).map_err(|e| (e, stream)) -} - -/// Due to different mutability requirements between Unix and Windows, -/// we are forced to clone the stream on Windows. -/// -/// # Errors -/// -/// As of winpipe 0.1.1, [`WinStream::try_clone`] is actually infallible. -#[cfg(windows)] -fn command(stream: Stream, handle: H) -> Result<(), (CommandError, Stream)> -where - H: Handle + 'static, - H::Sessions: serde::Serialize, - CommandResult: From, - E: serde::Serialize, -{ - let mut reader = match stream.try_clone() { - Ok(reader) => reader, - Err(err) => return Err((err.into(), stream)), - }; - let reader = BufReader::new(&mut reader); - - let mut writer = match stream.try_clone() { - Ok(writer) => writer, - Err(err) => return Err((err.into(), stream)), - }; - let writer = LineWriter::new(&mut writer); - - command_internal(reader, writer, handle).map_err(|e| (e, stream)) -} - -#[inline(always)] -fn command_internal( - mut reader: BufReader, - mut writer: LineWriter, - mut handle: H, -) -> Result<(), CommandError> -where - H: Handle + 'static, - H::Sessions: serde::Serialize, - CommandResult: From, - E: serde::Serialize, - R: io::Read, - W: io::Write, -{ let mut line = String::new(); reader.read_line(&mut line)?; @@ -319,7 +272,7 @@ mod tests { let handle = test::handle::Handle::default(); let socket = tmp.path().join("alice.sock"); let rids = test::arbitrary::set::(1..3); - let listener = Listener::bind(&socket).unwrap(); + let listener = UnixListener::bind(&socket).unwrap(); let nid = handle.nid().unwrap(); thread::spawn({ @@ -330,7 +283,7 @@ mod tests { for rid in &rids { let mut stream = loop { - if let Ok(stream) = Stream::connect(&socket) { + if let Ok(stream) = UnixStream::connect(&socket) { break stream; } }; @@ -369,7 +322,7 @@ mod tests { let socket = tmp.path().join("node.sock"); let proj = test::arbitrary::gen::(1); let peer = test::arbitrary::gen::(1); - let listener = Listener::bind(&socket).unwrap(); + let listener = UnixListener::bind(&socket).unwrap(); let mut handle = Node::new(&socket); thread::spawn({ diff --git a/crates/radicle-node/src/runtime.rs b/crates/radicle-node/src/runtime.rs index 115ef587..d1b1befe 100644 --- a/crates/radicle-node/src/runtime.rs +++ b/crates/radicle-node/src/runtime.rs @@ -6,9 +6,9 @@ use std::path::PathBuf; use std::{fs, io, net}; #[cfg(unix)] -use std::os::unix::net::UnixListener as Listener; +use std::os::unix::net::UnixListener; #[cfg(windows)] -use winpipe::WinListener as Listener; +use uds_windows::UnixListener; use crossbeam_channel as chan; use cyphernet::Ecdh; @@ -99,12 +99,12 @@ impl From for Error { } } -/// Wraps a [`Listener`] but tracks its origin. +/// Wraps a [`UnixListener`] but tracks its origin. pub enum ControlSocket { /// The listener was created by binding to it. - Bound(Listener, PathBuf), + Bound(UnixListener, PathBuf), /// The listener was received via socket activation. - Received(Listener), + Received(UnixListener), } /// Holds join handles to the client threads, as well as a client handle. @@ -323,7 +323,7 @@ impl Runtime { } #[cfg(all(feature = "systemd", target_os = "linux"))] - fn receive_listener() -> Option { + fn receive_listener() -> Option { let fd = match radicle_systemd::listen::fd("control") { Ok(Some(fd)) => fd, Ok(None) => return None, @@ -348,7 +348,7 @@ impl Runtime { return None; } - Some(Listener::from(socket)) + Some(UnixListener::from(socket)) } fn bind(path: PathBuf) -> Result { @@ -361,7 +361,7 @@ impl Runtime { } log::info!(target: "node", "Binding control socket {}..", &path.display()); - match Listener::bind(&path) { + match UnixListener::bind(&path) { Ok(sock) => Ok(ControlSocket::Bound(sock, path)), Err(err) if err.kind() == io::ErrorKind::AddrInUse => Err(Error::AlreadyRunning(path)), Err(err) => Err(err.into()), diff --git a/crates/radicle-node/src/runtime/handle.rs b/crates/radicle-node/src/runtime/handle.rs index d866bf07..d092fada 100644 --- a/crates/radicle-node/src/runtime/handle.rs +++ b/crates/radicle-node/src/runtime/handle.rs @@ -5,9 +5,9 @@ use std::sync::Arc; use std::{fmt, io, time}; #[cfg(unix)] -use std::os::unix::net::UnixStream as Stream; +use std::os::unix::net::UnixStream; #[cfg(windows)] -use winpipe::WinStream as Stream; +use uds_windows::UnixStream; use crossbeam_channel as chan; use radicle::crypto::PublicKey; @@ -338,7 +338,7 @@ impl radicle::node::Handle for Handle { // Send a shutdown request to our own control socket. This is the only way to kill the // control thread gracefully. Since the control thread may have called this function, // the control socket may already be disconnected. Ignore errors. - Stream::connect(self.home.socket()) + UnixStream::connect(self.home.socket()) .and_then(|sock| Command::Shutdown.to_writer(sock)) .ok(); diff --git a/crates/radicle/Cargo.toml b/crates/radicle/Cargo.toml index b33f07ca..6437a69f 100644 --- a/crates/radicle/Cargo.toml +++ b/crates/radicle/Cargo.toml @@ -63,7 +63,7 @@ unicode-normalization = { version = "0.1" } libc = { workspace = true } [target.'cfg(windows)'.dependencies] -winpipe = { workspace = true } +uds_windows = { workspace = true } [dev-dependencies] emojis = "0.6" diff --git a/crates/radicle/src/node.rs b/crates/radicle/src/node.rs index c9623890..2dafb663 100644 --- a/crates/radicle/src/node.rs +++ b/crates/radicle/src/node.rs @@ -25,9 +25,9 @@ use std::str::FromStr; use std::{fmt, io, net, thread, time}; #[cfg(unix)] -use std::os::unix::net::UnixStream as Stream; +use std::os::unix::net::UnixStream; #[cfg(windows)] -use winpipe::WinStream as Stream; +use uds_windows::UnixStream; use amplify::WrapperMut; use cyphernet::addr::NetAddr; @@ -948,7 +948,7 @@ pub trait Handle: Clone + Sync + Send { /// The iterator blocks for a `timeout` duration, returning [`Error::TimedOut`] /// if the duration is reached. pub struct LineIter { - stream: BufReader, + stream: BufReader, timeout: time::Duration, witness: PhantomData, } @@ -1009,7 +1009,7 @@ impl Node { cmd: Command, timeout: time::Duration, ) -> Result, Error> { - let mut stream = Stream::connect(&self.socket) + let mut stream = UnixStream::connect(&self.socket) .map_err(|e| Error::Connect(self.socket.clone(), e.kind()))?; cmd.to_writer(&mut stream)?; Ok(LineIter { diff --git a/crates/radicle/src/profile.rs b/crates/radicle/src/profile.rs index fcb5f8a7..79488aaa 100644 --- a/crates/radicle/src/profile.rs +++ b/crates/radicle/src/profile.rs @@ -603,32 +603,11 @@ impl Home { } pub fn socket(&self) -> PathBuf { - use env::RAD_SOCKET; - - #[cfg(unix)] const DEFAULT_SOCKET_NAME: &str = "control.sock"; - #[cfg(windows)] - const DEFAULT_SOCKET_NAME: &str = r#"\\.\pipe\radicle-node"#; - - match env::var_os(RAD_SOCKET).map(PathBuf::from) { - None => { - #[cfg(unix)] - return self.node().join(DEFAULT_SOCKET_NAME); - - #[cfg(windows)] - return PathBuf::from(DEFAULT_SOCKET_NAME); - } - Some(path) => { - #[cfg(windows)] - { - const PIPE_PREFIX: &str = r#"\\.\pipe\"#; - assert!(path.starts_with(PIPE_PREFIX), "The value of the environment variable {RAD_SOCKET} ('{}') must start with {PIPE_PREFIX}. This restriction might be relaxed in the future.", path.display()); - } - - path - } - } + env::var_os(env::RAD_SOCKET) + .map(PathBuf::from) + .unwrap_or_else(|| self.node().join(DEFAULT_SOCKET_NAME)) } /// Return a read-write handle to the notifications database.