From 19a262d3d67ea92a37ceaf18e79a783d8e65ccdd Mon Sep 17 00:00:00 2001 From: Lorenz Leutgeb Date: Thu, 21 Aug 2025 18:16:03 +0200 Subject: [PATCH] node: Use winpipe for control socket on Windows As `std::os::unix` is obviously not available on Windows, resort to the `winpipe` crate, which implements a very similar API for named pipes. Apart from simple changes to imports, we need to jump through a few hoops because the `std::io::{Read, Write}` impls are on `&mut` for `winpipe`, thus yielding different mutability requirements on Unix vs. Windows. We resort to cloning, but as this fallible, swallow the added complexity of a platform-dependent implementation to not risk panics on Unix. --- Cargo.lock | 1 + crates/radicle-node/Cargo.toml | 3 ++ crates/radicle-node/src/control.rs | 66 ++++++++++++++++++----- crates/radicle-node/src/runtime.rs | 16 +++--- crates/radicle-node/src/runtime/handle.rs | 8 ++- 5 files changed, 74 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 36dc6177..01711a6c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2878,6 +2878,7 @@ dependencies = [ "tempfile", "test-log", "thiserror 1.0.69", + "winpipe", ] [[package]] diff --git a/crates/radicle-node/Cargo.toml b/crates/radicle-node/Cargo.toml index e5fea061..667b084c 100644 --- a/crates/radicle-node/Cargo.toml +++ b/crates/radicle-node/Cargo.toml @@ -51,6 +51,9 @@ thiserror = { workspace = true } [target.'cfg(target_os = "linux")'.dependencies] radicle-systemd = { workspace = true, optional = true } +[target.'cfg(windows)'.dependencies] +winpipe = { workspace = true } + [dev-dependencies] qcheck = { workspace = true } qcheck-macros = { workspace = true } diff --git a/crates/radicle-node/src/control.rs b/crates/radicle-node/src/control.rs index 0f40b74d..c2dfcffb 100644 --- a/crates/radicle-node/src/control.rs +++ b/crates/radicle-node/src/control.rs @@ -2,11 +2,14 @@ use std::io::prelude::*; use std::io::BufReader; use std::io::LineWriter; -use std::os::unix::net::UnixListener; -use std::os::unix::net::UnixStream; use std::path::PathBuf; use std::{io, net, time}; +#[cfg(unix)] +use std::os::unix::net::{UnixListener as Listener, UnixStream as Stream}; +#[cfg(windows)] +use winpipe::{WinListener as Listener, WinStream as Stream}; + use radicle::node::Handle; use serde_json as json; @@ -30,7 +33,7 @@ pub enum Error { } /// Listen for commands on the control socket, and process them. -pub fn listen(listener: UnixListener, handle: H) -> Result<(), Error> +pub fn listen(listener: Listener, handle: H) -> Result<(), Error> where H: Handle + 'static, H::Sessions: serde::Serialize, @@ -42,11 +45,11 @@ where for incoming in listener.incoming() { match incoming { - Ok(mut stream) => { + Ok(stream) => { let handle = handle.clone(); thread::spawn(&nid, "control", move || { - if let Err(e) = command(&stream, handle) { + if let Err((e, mut stream)) = command(stream, handle) { log::error!(target: "control", "Command returned error: {e}"); CommandResult::error(e).to_writer(&mut stream).ok(); @@ -74,15 +77,55 @@ enum CommandError { Io(#[from] io::Error), } -fn command(stream: &UnixStream, mut handle: H) -> Result<(), CommandError> +#[cfg(unix)] +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 = BufReader::new(stream); - let mut writer = LineWriter::new(stream); + let reader = BufReader::new(&stream); + let 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(mut stream: Stream, handle: H) -> Result<(), (CommandError, Stream)> +where + H: Handle + 'static, + H::Sessions: serde::Serialize, + CommandResult: From, + E: serde::Serialize, +{ + let mut clone = stream.try_clone().map_err(|e| (e.into(), stream))?; + let reader = BufReader::new(&mut clone); + let writer = LineWriter::new(&mut stream); + + 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)?; @@ -241,7 +284,6 @@ fn fetch>( #[cfg(test)] mod tests { use std::io::prelude::*; - use std::os::unix::net::UnixStream; use std::thread; use super::*; @@ -257,7 +299,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 = UnixListener::bind(&socket).unwrap(); + let listener = Listener::bind(&socket).unwrap(); thread::spawn({ let handle = handle.clone(); @@ -267,7 +309,7 @@ mod tests { for rid in &rids { let stream = loop { - if let Ok(stream) = UnixStream::connect(&socket) { + if let Ok(stream) = Stream::connect(&socket) { break stream; } }; @@ -305,7 +347,7 @@ mod tests { let socket = tmp.path().join("node.sock"); let proj = test::arbitrary::gen::(1); let peer = test::arbitrary::gen::(1); - let listener = UnixListener::bind(&socket).unwrap(); + let listener = Listener::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 d50ceb0f..90e33d48 100644 --- a/crates/radicle-node/src/runtime.rs +++ b/crates/radicle-node/src/runtime.rs @@ -1,10 +1,14 @@ pub mod handle; pub mod thread; -use std::os::unix::net::UnixListener; use std::path::PathBuf; use std::{fs, io, net}; +#[cfg(unix)] +use std::os::unix::net::UnixListener as Listener; +#[cfg(windows)] +use winpipe::WinListener as Listener; + use crossbeam_channel as chan; use cyphernet::Ecdh; use netservices::resource::NetAccept; @@ -99,9 +103,9 @@ impl From for Error { /// Wraps a [`UnixListener`] but tracks its origin. pub enum ControlSocket { /// The listener was created by binding to it. - Bound(UnixListener, PathBuf), + Bound(Listener, PathBuf), /// The listener was received via socket activation. - Received(UnixListener), + Received(Listener), } /// Holds join handles to the client threads, as well as a client handle. @@ -316,7 +320,7 @@ impl Runtime { } #[cfg(all(feature = "systemd", target_os = "linux"))] - fn receive_listener() -> Option { + fn receive_listener() -> Option { use std::os::fd::FromRawFd; match radicle_systemd::listen::fd("control") { Ok(Some(fd)) => { @@ -330,7 +334,7 @@ impl Runtime { Some(unsafe { // SAFETY: We take ownership of this FD from systemd, // which guarantees that it is open. - UnixListener::from_raw_fd(fd) + Listener::from_raw_fd(fd) }) } Ok(None) => None, @@ -351,7 +355,7 @@ impl Runtime { } log::info!(target: "node", "Binding control socket {}..", &path.display()); - match UnixListener::bind(&path) { + match Listener::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 f93f61cb..d071308f 100644 --- a/crates/radicle-node/src/runtime/handle.rs +++ b/crates/radicle-node/src/runtime/handle.rs @@ -1,9 +1,13 @@ use std::net; -use std::os::unix::net::UnixStream; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::{fmt, io, time}; +#[cfg(unix)] +use std::os::unix::net::UnixStream as Stream; +#[cfg(windows)] +use winpipe::WinStream as Stream; + use crossbeam_channel as chan; use radicle::node::events::{Event, Events}; use radicle::node::policy; @@ -320,7 +324,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. - UnixStream::connect(self.home.socket()) + Stream::connect(self.home.socket()) .and_then(|sock| Command::Shutdown.to_writer(sock)) .ok();