diff --git a/README.md b/README.md index d76dab2..ec79572 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ geth cas add ./file The daemon owns local identity, the Iroh endpoint, trust state, resource registry, module router, local metadata store, and synchronized data structures. Most non-daemon commands talk to the daemon through a local Unix socket at -`$GETH_HOME/run/geth.sock`. +`$GETH_HOME/run/geth.sock` on Linux/macOS or a per-home Windows named pipe. `geth status` and `geth status --json` report daemon uptime, store schema and durability settings, Iroh endpoint/relay/discovery state, and native backend health for automation. @@ -472,7 +472,7 @@ $GETH_HOME/ identity/agent.ed25519 identity/iroh.ed25519 cas/blobs/ - run/geth.sock + run/geth.sock # Linux/macOS local control; Windows uses a named pipe ``` ## Quick Start @@ -514,7 +514,7 @@ empty target home for validation. It refuses to overwrite a non-empty target. ## Doctor `geth doctor` is a local operational check that works even when the daemon is -not reachable. It checks config parsing, local daemon socket health, +not reachable. It checks config parsing, local daemon control-endpoint health, `ssh-keygen` availability, metadata-store readability, imported peer-card validity, and representative peer grants when local metadata is available. Use `--json` for scripts. diff --git a/crates/geth-config/src/lib.rs b/crates/geth-config/src/lib.rs index 64dc08f..d1f7539 100644 --- a/crates/geth-config/src/lib.rs +++ b/crates/geth-config/src/lib.rs @@ -87,7 +87,23 @@ impl GethPaths { #[must_use] pub fn socket_path(&self) -> PathBuf { - self.run_dir().join("geth.sock") + PathBuf::from(self.control_endpoint()) + } + + #[must_use] + pub fn control_endpoint(&self) -> String { + #[cfg(unix)] + { + self.run_dir().join("geth.sock").display().to_string() + } + #[cfg(windows)] + { + let home = self.home.display().to_string().to_ascii_lowercase(); + format!( + r"\\.\pipe\geth-{:016x}", + stable_control_endpoint_suffix(home.as_bytes()) + ) + } } pub fn ensure_base_dirs(&self) -> Result<(), ConfigError> { @@ -98,6 +114,16 @@ impl GethPaths { } } +#[cfg(any(windows, test))] +fn stable_control_endpoint_suffix(bytes: &[u8]) -> u64 { + let mut hash = 0xcbf2_9ce4_8422_2325_u64; + for byte in bytes { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + hash +} + #[derive(Debug, thiserror::Error)] pub enum ConfigError { #[error("could not determine OS data directory and GETH_HOME is unset")] @@ -603,6 +629,18 @@ mod tests { assert!(matches!(error, ConfigError::InvalidLocalDiscovery(_))); } + #[test] + fn control_endpoint_suffix_is_stable_and_home_scoped() { + assert_eq!( + stable_control_endpoint_suffix(b"C:/Users/Eric/geth"), + stable_control_endpoint_suffix(b"C:/Users/Eric/geth") + ); + assert_ne!( + stable_control_endpoint_suffix(b"C:/Users/Eric/geth"), + stable_control_endpoint_suffix(b"C:/Users/Eric/other") + ); + } + #[test] fn config_set_preserves_comments_and_validates_the_result() { let home = tempfile::tempdir().expect("tempdir"); diff --git a/crates/geth-node/src/daemon.rs b/crates/geth-node/src/daemon.rs index 71dcc0c..c76b6c6 100644 --- a/crates/geth-node/src/daemon.rs +++ b/crates/geth-node/src/daemon.rs @@ -1,20 +1,20 @@ //! Daemon task orchestration helpers. use crate::{ - LocalNode, NodeError, handle_stream, init_node, mirror_all_kv_stores_to_iroh_docs, + LocalNode, NodeError, handle_stream, init_node, local_transport, + local_transport::LocalListener, mirror_all_kv_stores_to_iroh_docs, mirror_local_cas_to_iroh_blobs, start_peer_card_lan_discovery, sync::run_live_sync_once, }; use geth_config::{GethConfig, GethPaths, RelayMode}; use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode}; use geth_store::Store; use std::time::Duration; -use tokio::net::{UnixListener, UnixStream}; use tokio::task::JoinHandle; pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> { let mut node = init_node(&paths)?; let listener = bind_control_socket(&paths).await?; - let _socket_guard = ControlSocketGuard { + let _control_guard = ControlEndpointGuard { paths: paths.clone(), }; let config = GethConfig::load(&node.paths.config_file())?; @@ -32,7 +32,7 @@ pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> { let serve_result = async { let _peer_card_lan_discovery = start_peer_card_lan_discovery(&node).await; - tracing::info!(socket = %paths.socket_path().display(), "geth daemon listening"); + tracing::info!(endpoint = %paths.control_endpoint(), "geth daemon listening"); serve_local_control(node.clone(), listener).await } .await; @@ -47,27 +47,20 @@ pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> { Ok(()) } -struct ControlSocketGuard { +struct ControlEndpointGuard { paths: GethPaths, } -impl Drop for ControlSocketGuard { +impl Drop for ControlEndpointGuard { fn drop(&mut self) { if let Err(error) = remove_control_socket(&self.paths) { - tracing::warn!(%error, "failed to clean up daemon control socket"); + tracing::warn!(%error, "failed to clean up daemon control endpoint"); } } } -async fn bind_control_socket(paths: &GethPaths) -> Result { - let socket_path = paths.socket_path(); - if socket_path.exists() { - match UnixStream::connect(&socket_path).await { - Ok(_) => return Err(NodeError::DaemonAlreadyRunning(paths.home().to_path_buf())), - Err(_) => remove_control_socket(paths)?, - } - } - match UnixListener::bind(&socket_path) { +async fn bind_control_socket(paths: &GethPaths) -> Result { + match local_transport::bind(paths).await { Ok(listener) => Ok(listener), Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => { Err(NodeError::DaemonAlreadyRunning(paths.home().to_path_buf())) @@ -77,11 +70,7 @@ async fn bind_control_socket(paths: &GethPaths) -> Result Result<(), NodeError> { - match std::fs::remove_file(paths.socket_path()) { - Ok(()) => Ok(()), - Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), - Err(error) => Err(error.into()), - } + local_transport::remove(paths).map_err(NodeError::from) } async fn shutdown_iroh_runtime( @@ -115,11 +104,14 @@ fn clear_iroh_runtime_handles(node: &LocalNode) -> Result<(), NodeError> { Ok(()) } -async fn serve_local_control(node: LocalNode, listener: UnixListener) -> Result<(), NodeError> { +async fn serve_local_control( + node: LocalNode, + mut listener: LocalListener, +) -> Result<(), NodeError> { loop { tokio::select! { accepted = listener.accept() => { - let (stream, _) = accepted?; + let stream = accepted?; let node = node.clone(); tokio::spawn(async move { if let Err(error) = handle_stream(node, stream).await { @@ -313,7 +305,45 @@ fn config_relay_mode_to_iroh( mod tests { use super::*; + #[tokio::test] + async fn local_control_roundtrip_uses_the_platform_transport() { + let dir = tempfile::tempdir().expect("tempdir"); + let paths = GethPaths::from_home(dir.path()); + let node = init_node(&paths).expect("init node"); + let mut listener = match bind_control_socket(&paths).await { + Ok(listener) => listener, + #[cfg(unix)] + Err(NodeError::Io(error)) if error.kind() == std::io::ErrorKind::PermissionDenied => { + eprintln!("skipping local control roundtrip; Unix sockets unavailable: {error}"); + return; + } + Err(error) => panic!("bind control: {error}"), + }; + let server = tokio::spawn(async move { + for _ in 0..2 { + let stream = listener.accept().await.expect("accept control"); + if handle_stream(node.clone(), stream).await.is_ok() { + return; + } + } + panic!("local control server did not receive a valid request"); + }); + + let second = bind_control_socket(&paths) + .await + .expect_err("second daemon must not replace the live endpoint"); + assert!(matches!(second, NodeError::DaemonAlreadyRunning(_))); + + let response = crate::send_control(&paths, geth_control::ControlRequest::Status) + .await + .expect("send status"); + assert!(matches!(response, geth_control::ControlResponse::Status(_))); + server.await.expect("join control server"); + remove_control_socket(&paths).expect("cleanup control endpoint"); + } + #[test] + #[cfg(unix)] fn control_socket_cleanup_is_idempotent() { let dir = tempfile::tempdir().expect("tempdir"); let paths = GethPaths::from_home(dir.path()); @@ -410,6 +440,7 @@ mod tests { } #[tokio::test] + #[cfg(unix)] async fn control_socket_binding_rejects_a_second_daemon_and_recovers_stale_paths() { let dir = tempfile::tempdir().expect("tempdir"); let paths = GethPaths::from_home(dir.path()); diff --git a/crates/geth-node/src/doctor.rs b/crates/geth-node/src/doctor.rs index d8aaf7c..7388dc2 100644 --- a/crates/geth-node/src/doctor.rs +++ b/crates/geth-node/src/doctor.rs @@ -56,15 +56,6 @@ fn check_config(paths: &GethPaths, checks: &mut Vec) { } async fn check_daemon(paths: &GethPaths, checks: &mut Vec) { - if !paths.socket_path().exists() { - checks.push(fail( - "daemon-not-running", - format!("daemon socket is absent: {}", paths.socket_path().display()), - "start the daemon with `geth daemon run` or the user service command for your platform", - )); - return; - } - match send_control(paths, ControlRequest::Status).await { Ok(ControlResponse::Status(status)) => checks.push(ok( "daemon-ok", @@ -79,13 +70,21 @@ async fn check_daemon(paths: &GethPaths, checks: &mut Vec) { format!("daemon returned unexpected response: {other:?}"), "restart the daemon and rerun `geth doctor`", )), - Err(error) => checks.push(fail( - "daemon-stale-socket", + Err(error) if crate::local_transport::endpoint_file_exists(paths) => checks.push(fail( + "daemon-stale-control-endpoint", format!( - "daemon socket exists but control connection failed: {}", + "daemon control endpoint exists but the connection failed: {}", error ), - "stop any stale daemon, remove the socket if no daemon is running, then start `geth daemon run`", + "restart the user service or foreground daemon; on Unix, remove the stale socket only after confirming no daemon is running", + )), + Err(error) => checks.push(fail( + "daemon-not-running", + format!( + "daemon control endpoint did not answer at {}: {error}", + paths.control_endpoint() + ), + "start the daemon with `geth daemon run` or `geth daemon install`", )), } } @@ -320,6 +319,7 @@ mod tests { } #[tokio::test] + #[cfg(unix)] async fn doctor_reports_bad_config_and_stale_socket() { let home = tempfile::tempdir().expect("home"); let paths = GethPaths::from_home(home.path()); @@ -329,7 +329,7 @@ mod tests { let report = run_doctor(&paths).await.expect("doctor"); assert!(!report.ok); assert!(has_code(&report, "config-invalid")); - assert!(has_code(&report, "daemon-stale-socket")); + assert!(has_code(&report, "daemon-stale-control-endpoint")); } fn has_code(report: &DoctorReport, code: &str) -> bool { diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 0fd3429..5ef848b 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -2,6 +2,7 @@ pub mod backup; mod daemon; pub mod doctor; mod local_control; +mod local_transport; mod peer_client; mod peer_control; mod resource_contracts; @@ -68,6 +69,7 @@ use geth_types::{ use iroh::protocol::ProtocolHandler; use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode}; pub use local_control::handle_request_async; +use local_transport::LocalStream; use peer_client::{request_overlay_wire, request_peer_control, request_pipe_wire}; use peer_control::{ IrohAlpnRoute, PeerControlCaller, authenticate_peer_control_caller, classify_iroh_alpn, @@ -85,7 +87,9 @@ use sync::{ KvDocsState, load_live_sync_cursor, store_live_sync_cursor, sync_now, sync_status_local, }; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream}; +use tokio::net::{TcpListener, TcpStream}; +#[cfg(unix)] +use tokio::net::{UnixListener, UnixStream}; use tokio::sync::{mpsc, oneshot}; use tun_rs::{DeviceBuilder, Layer}; use wire::{ @@ -429,7 +433,7 @@ pub async fn send_control( paths: &GethPaths, request: ControlRequest, ) -> Result { - let mut stream = UnixStream::connect(paths.socket_path()).await?; + let mut stream = local_transport::connect(paths).await?; stream .write_all(geth_control::encode_request(&request)?.as_bytes()) .await?; @@ -445,7 +449,7 @@ pub async fn stream_ssh_proxy( peer_node: String, bearer_secret: Option, ) -> Result<(), NodeError> { - let mut stream = UnixStream::connect(paths.socket_path()).await?; + let mut stream = local_transport::connect(paths).await?; stream .write_all( geth_control::encode_request(&ControlRequest::SshProxyStream { @@ -461,7 +465,7 @@ pub async fn stream_ssh_proxy( match geth_control::decode_response(&line)? { ControlResponse::SshProxyConnected { allowed: true, .. } => { let stream = reader.into_inner(); - let (mut local_read, mut local_write) = stream.into_split(); + let (mut local_read, mut local_write) = tokio::io::split(stream); let mut stdin = tokio::io::stdin(); let mut stdout = tokio::io::stdout(); let upload = async { @@ -526,7 +530,7 @@ async fn stream_pipe_tcp( bearer_secret: Option, client: TcpStream, ) -> Result<(), NodeError> { - let mut stream = UnixStream::connect(paths.socket_path()).await?; + let mut stream = local_transport::connect(paths).await?; stream .write_all( geth_control::encode_request(&ControlRequest::PipeTcpStream { @@ -543,7 +547,7 @@ async fn stream_pipe_tcp( match geth_control::decode_response(&line)? { ControlResponse::PipeRemoteConnected { allowed: true, .. } => { let daemon_stream = reader.into_inner(); - let (mut daemon_read, mut daemon_write) = daemon_stream.into_split(); + let (mut daemon_read, mut daemon_write) = tokio::io::split(daemon_stream); let (mut client_read, mut client_write) = client.into_split(); let upload = async { tokio::io::copy(&mut client_read, &mut daemon_write).await?; @@ -568,6 +572,7 @@ async fn stream_pipe_tcp( } } +#[cfg(unix)] pub async fn run_unix_forward( paths: &GethPaths, listen_path: PathBuf, @@ -605,6 +610,20 @@ pub async fn run_unix_forward( } } +#[cfg(not(unix))] +pub async fn run_unix_forward( + _paths: &GethPaths, + _listen_path: PathBuf, + _peer_node: String, + _target_path: PathBuf, + _bearer_secret: Option, +) -> Result<(), NodeError> { + Err(NodeError::UnsupportedPlatform( + geth_pipe::unix_forward_unsupported_message().to_owned(), + )) +} + +#[cfg(unix)] async fn stream_pipe_unix( paths: &GethPaths, peer_node: String, @@ -612,7 +631,7 @@ async fn stream_pipe_unix( bearer_secret: Option, client: UnixStream, ) -> Result<(), NodeError> { - let mut stream = UnixStream::connect(paths.socket_path()).await?; + let mut stream = local_transport::connect(paths).await?; stream .write_all( geth_control::encode_request(&ControlRequest::PipeUnixStream { @@ -629,7 +648,7 @@ async fn stream_pipe_unix( match geth_control::decode_response(&line)? { ControlResponse::PipeRemoteConnected { allowed: true, .. } => { let daemon_stream = reader.into_inner(); - let (mut daemon_read, mut daemon_write) = daemon_stream.into_split(); + let (mut daemon_read, mut daemon_write) = tokio::io::split(daemon_stream); let (mut client_read, mut client_write) = client.into_split(); let upload = async { tokio::io::copy(&mut client_read, &mut daemon_write).await?; @@ -654,7 +673,7 @@ async fn stream_pipe_unix( } } -async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeError> { +async fn handle_stream(node: LocalNode, stream: LocalStream) -> Result<(), NodeError> { let mut reader = BufReader::new(stream); let mut line = String::new(); reader.read_line(&mut line).await?; @@ -737,6 +756,7 @@ async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeEr } return result; } + #[cfg(unix)] if let ControlRequest::PipeUnixStream { node: peer_node, target_path, @@ -2661,7 +2681,7 @@ async fn handle_local_pipe_tcp_stream( peer_node: &str, target_addr: String, bearer_secret: Option, - local_stream: UnixStream, + local_stream: LocalStream, ) -> Result<(), NodeError> { geth_pipe::validate_tcp_forward_target_addr(&target_addr)?; let store = Store::open(&node.paths.metadata_db())?; @@ -2753,7 +2773,7 @@ async fn handle_local_pipe_tcp_stream( let ControlResponse::PipeRemoteConnected { allowed: true, .. } = local_response else { return Ok(()); }; - let (mut local_read, mut local_write) = local_stream.into_split(); + let (mut local_read, mut local_write) = tokio::io::split(local_stream); let upload = async { tokio::io::copy(&mut local_read, &mut remote_send) .await @@ -2773,12 +2793,13 @@ async fn handle_local_pipe_tcp_stream( Ok(()) } +#[cfg(unix)] async fn handle_local_pipe_unix_stream( node: LocalNode, peer_node: &str, target_path: PathBuf, bearer_secret: Option, - local_stream: UnixStream, + local_stream: LocalStream, ) -> Result<(), NodeError> { let target_path = geth_pipe::validate_unix_forward_path(&target_path)?; let target_display = target_path.display().to_string(); @@ -2871,7 +2892,7 @@ async fn handle_local_pipe_unix_stream( let ControlResponse::PipeRemoteConnected { allowed: true, .. } = local_response else { return Ok(()); }; - let (mut local_read, mut local_write) = local_stream.into_split(); + let (mut local_read, mut local_write) = tokio::io::split(local_stream); let upload = async { tokio::io::copy(&mut local_read, &mut remote_send) .await @@ -2988,7 +3009,7 @@ async fn handle_local_ssh_proxy_stream( node: LocalNode, peer_node: &str, bearer_secret: Option, - local_stream: UnixStream, + local_stream: LocalStream, ) -> Result<(), NodeError> { let store = Store::open(&node.paths.metadata_db())?; let stored = store @@ -3076,7 +3097,7 @@ async fn handle_local_ssh_proxy_stream( let ControlResponse::SshProxyConnected { allowed: true, .. } = local_response else { return Ok(()); }; - let (mut local_read, mut local_write) = local_stream.into_split(); + let (mut local_read, mut local_write) = tokio::io::split(local_stream); let upload = async { tokio::io::copy(&mut local_read, &mut remote_send) .await @@ -5519,6 +5540,7 @@ async fn handle_pipe_tcp_wire_connection( Ok(()) } +#[cfg(unix)] async fn handle_pipe_unix_wire_connection( node: LocalNode, remote_endpoint_id: &str, @@ -5620,6 +5642,24 @@ async fn handle_pipe_unix_wire_connection( Ok(()) } +#[cfg(not(unix))] +async fn handle_pipe_unix_wire_connection( + _node: LocalNode, + _remote_endpoint_id: &str, + _request: PipeUnixConnectWire, + mut send: iroh::endpoint::SendStream, + _recv: iroh::endpoint::RecvStream, +) -> Result<(), NodeError> { + let response = PipeWireResponse::Error { + message: geth_pipe::unix_forward_unsupported_message().to_owned(), + }; + send.write_all(geth_control::encode_pipe_wire_response(&response)?.as_bytes()) + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + send.finish() + .map_err(|error| NodeError::IrohPeer(error.to_string())) +} + fn peer_gossip_endpoint_id( node: &LocalNode, peer_node: &str, @@ -12926,7 +12966,7 @@ mod tests { socket.write_all(b"pong").await.expect("write tcp echo"); socket.shutdown().await.expect("shutdown tcp echo"); }); - let (forward_client, forward_daemon) = UnixStream::pair().expect("unix stream pair"); + let (forward_client, forward_daemon) = tokio::io::duplex(64 * 1024); let right_node_id = right_card.node_id.to_string(); let left_for_forward = left.clone(); let tcp_target_for_forward = tcp_target.clone(); @@ -12936,7 +12976,7 @@ mod tests { &right_node_id, tcp_target_for_forward, None, - forward_daemon, + Box::new(forward_daemon), ) .await }); @@ -12958,7 +12998,7 @@ mod tests { other => panic!("unexpected pipe TCP stream response: {other:?}"), } let stream = forward_reader.into_inner(); - let (mut forward_read, mut forward_write) = stream.into_split(); + let (mut forward_read, mut forward_write) = tokio::io::split(stream); forward_write .write_all(b"ping") .await @@ -13015,7 +13055,7 @@ mod tests { &right_node_id, unix_target_for_forward, None, - unix_forward_daemon, + Box::new(unix_forward_daemon), ) .await }); diff --git a/crates/geth-node/src/local_transport.rs b/crates/geth-node/src/local_transport.rs new file mode 100644 index 0000000..fed9fc1 --- /dev/null +++ b/crates/geth-node/src/local_transport.rs @@ -0,0 +1,140 @@ +use geth_config::GethPaths; +use std::io; +use tokio::io::{AsyncRead, AsyncWrite}; + +pub(crate) trait LocalIo: AsyncRead + AsyncWrite + Unpin + Send {} + +impl LocalIo for T where T: AsyncRead + AsyncWrite + Unpin + Send {} + +pub(crate) type LocalStream = Box; + +#[cfg(unix)] +#[derive(Debug)] +pub(crate) struct LocalListener(tokio::net::UnixListener); + +#[cfg(unix)] +pub(crate) async fn bind(paths: &GethPaths) -> io::Result { + let socket_path = paths.socket_path(); + if socket_path.exists() { + match tokio::net::UnixStream::connect(&socket_path).await { + Ok(_) => { + return Err(io::Error::new( + io::ErrorKind::AddrInUse, + "geth local control socket is already accepting connections", + )); + } + Err(_) => remove(paths)?, + } + } + tokio::net::UnixListener::bind(socket_path).map(LocalListener) +} + +#[cfg(unix)] +pub(crate) async fn connect(paths: &GethPaths) -> io::Result { + Ok(Box::new( + tokio::net::UnixStream::connect(paths.socket_path()).await?, + )) +} + +#[cfg(unix)] +impl LocalListener { + pub(crate) async fn accept(&mut self) -> io::Result { + let (stream, _) = self.0.accept().await?; + Ok(Box::new(stream)) + } +} + +#[cfg(unix)] +pub(crate) fn remove(paths: &GethPaths) -> io::Result<()> { + match std::fs::remove_file(paths.socket_path()) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +#[cfg(unix)] +pub(crate) fn endpoint_file_exists(paths: &GethPaths) -> bool { + paths.socket_path().exists() +} + +#[cfg(windows)] +#[derive(Debug)] +pub(crate) struct LocalListener { + pipe_name: String, + pending: Option, +} + +#[cfg(windows)] +pub(crate) async fn bind(paths: &GethPaths) -> io::Result { + use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions}; + + let pipe_name = paths.control_endpoint(); + if ClientOptions::new().open(&pipe_name).is_ok() { + return Err(io::Error::new( + io::ErrorKind::AddrInUse, + "geth local control named pipe is already accepting connections", + )); + } + let pending = ServerOptions::new() + .first_pipe_instance(true) + .create(&pipe_name) + .map_err(map_windows_bind_error)?; + Ok(LocalListener { + pipe_name, + pending: Some(pending), + }) +} + +#[cfg(windows)] +fn map_windows_bind_error(error: io::Error) -> io::Error { + if matches!( + error.kind(), + io::ErrorKind::PermissionDenied | io::ErrorKind::AlreadyExists + ) { + io::Error::new(io::ErrorKind::AddrInUse, error) + } else { + error + } +} + +#[cfg(windows)] +pub(crate) async fn connect(paths: &GethPaths) -> io::Result { + use tokio::net::windows::named_pipe::ClientOptions; + + let pipe_name = paths.control_endpoint(); + let mut last_error = None; + for _ in 0..20 { + match ClientOptions::new().open(&pipe_name) { + Ok(client) => return Ok(Box::new(client)), + Err(error) => last_error = Some(error), + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + Err(last_error.unwrap_or_else(|| { + io::Error::new(io::ErrorKind::NotFound, "geth local control pipe is absent") + })) +} + +#[cfg(windows)] +impl LocalListener { + pub(crate) async fn accept(&mut self) -> io::Result { + use tokio::net::windows::named_pipe::ServerOptions; + + let connected = self.pending.as_mut().expect("pending pipe instance"); + connected.connect().await?; + let connected = self.pending.take().expect("connected pipe instance"); + self.pending = Some(ServerOptions::new().create(&self.pipe_name)?); + Ok(Box::new(connected)) + } +} + +#[cfg(windows)] +pub(crate) fn remove(_paths: &GethPaths) -> io::Result<()> { + Ok(()) +} + +#[cfg(windows)] +pub(crate) fn endpoint_file_exists(_paths: &GethPaths) -> bool { + false +} diff --git a/docs/architecture.md b/docs/architecture.md index 059d2f7..89dc2ae 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -3,11 +3,11 @@ `geth` is a single-binary local-first mesh runtime. One executable provides both daemon mode and control mode. The daemon owns local identity, metadata storage, the shared Iroh endpoint, resource registry, module routing, and local control -socket. Control commands connect to the Unix socket and send typed JSONL -requests. +endpoint. Control commands connect to a per-home Unix socket on Linux/macOS or +a per-home named pipe on Windows and send typed JSONL requests. Within `geth-node`, daemon lifecycle code is separated from feature handlers: -`daemon.rs` owns `geth daemon run` startup, local socket binding, shutdown +`daemon.rs` owns `geth daemon run` startup, local endpoint binding, shutdown signal handling, Iroh endpoint startup, the Iroh accept loop, and background live-sync task spawning. `local_control.rs` owns async local `ControlRequest` routing, safe trace-field classification, and named peer/resource/local handler @@ -28,9 +28,11 @@ as stale and replaced. This keeps one daemon authoritative for each geth home and prevents a second process from unlinking the first daemon's control path. After local control stops accepting requests, `daemon.rs` cancels the Iroh accept and live-sync tasks, releases native docs/gossip/blob handles, calls -`Endpoint::close().await`, and removes the local control socket. The same -cleanup path runs when the serving loop returns an error, preventing stale -socket files and unclosed endpoint clones from becoming restart behavior. +`Endpoint::close().await`, and releases the local control endpoint. On Unix this +also removes the socket path; Windows named pipes disappear with their server +handles. The same cleanup path runs when the serving loop returns an error, +preventing stale endpoints and unclosed endpoint clones from becoming restart +behavior. Foreground daemons handle Ctrl-C, and Unix daemons also handle the SIGTERM used by user service managers, through this same graceful shutdown path. diff --git a/docs/compatibility.md b/docs/compatibility.md index dd1dab3..bdb1285 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -60,9 +60,10 @@ cover the subset intended as stable rather than every incidental field. ## Local Control JSONL -The local daemon control socket uses newline-delimited JSON request and response -messages from `geth-control`. This protocol is local-only and not a remote trust -boundary, but local automation may still rely on it. +The local daemon control endpoint uses newline-delimited JSON request and +response messages from `geth-control`. Its carrier is a Unix socket on +Linux/macOS and a per-home Windows named pipe. This protocol is local-only and +not a remote trust boundary, but local automation may still rely on it. Stable local-control variants follow the same compatibility rules as `--json` output. Unknown variants, malformed JSON, and messages without a trailing diff --git a/docs/roadmap.md b/docs/roadmap.md index 0739387..9e5dff7 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -22,8 +22,10 @@ For deployment-readiness work that cuts across feature areas, see - `[x]` Daemon startup claims the local control endpoint before starting network and background modules. - `[x]` A live endpoint rejects another daemon with the stable - `daemon_already_running` code instead of unlinking the first socket. - - `[x]` Stale, unreachable socket paths are recovered automatically. + `daemon_already_running` code instead of unlinking the first Unix socket or + replacing the first Windows named-pipe server. + - `[x]` Stale, unreachable Unix socket paths are recovered automatically; + Windows pipe lifetime is owned by the server handle. - `[x]` Tests cover stale recovery and live second-daemon rejection. - `[x]` Keep local control available when Iroh-native startup degrades. @@ -37,6 +39,17 @@ For deployment-readiness work that cuts across feature areas, see - `[x]` Tests inject a post-endpoint native-store failure and verify clean degradation where UDP endpoint binding is available. +- `[x]` Use a platform-native local control carrier. + Acceptance criteria: + - `[x]` Linux and macOS retain per-home Unix-domain sockets. + - `[x]` Windows derives a deterministic per-home named-pipe name and uses + Tokio named-pipe clients and server instances for the same JSONL protocol. + - `[x]` Unary control, SSH proxying, and TCP byte forwarding share one async + local-stream abstraction without adding another executable or remote + transport. + - `[x]` Unix-socket forwarding is cfg-gated with an explicit unsupported + result on Windows, and a platform transport roundtrip runs in CI tests. + - `[x]` Make startup modes and the daemon lifecycle discoverable. Acceptance criteria: - `[x]` Base and nested CLI help explain every command family instead of @@ -346,10 +359,11 @@ control, local CAS, service installation, and written architecture decisions. - `cargo run -p geth -- daemon run` starts one local daemon. - No `gethd` or `gethctl` binaries exist in the workspace. -- `[x]` Local daemon control socket. +- `[x]` Local daemon control endpoint. Acceptance criteria: - Control request/response types roundtrip through JSONL serialization. - `geth status` and `geth node id` talk to a running daemon. + - Unix sockets and Windows named pipes carry the same local protocol. - Control decoding treats input as untrusted and returns structured errors. - `[x]` Local metadata store and identity.