Add Windows named-pipe local control

This commit is contained in:
Eric Wendland 2026-07-18 17:05:38 +02:00
commit 3c84713472
9 changed files with 338 additions and 72 deletions

View file

@ -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.

View file

@ -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");

View file

@ -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<UnixListener, NodeError> {
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<LocalListener, NodeError> {
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<UnixListener, NodeErro
}
fn remove_control_socket(paths: &GethPaths) -> 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());

View file

@ -56,15 +56,6 @@ fn check_config(paths: &GethPaths, checks: &mut Vec<DoctorCheck>) {
}
async fn check_daemon(paths: &GethPaths, checks: &mut Vec<DoctorCheck>) {
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<DoctorCheck>) {
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 {

View file

@ -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<ControlResponse, 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(&request)?.as_bytes())
.await?;
@ -445,7 +449,7 @@ pub async fn stream_ssh_proxy(
peer_node: String,
bearer_secret: Option<String>,
) -> 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<String>,
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<String>,
) -> 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<String>,
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<String>,
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<String>,
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<String>,
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
});

View file

@ -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<T> LocalIo for T where T: AsyncRead + AsyncWrite + Unpin + Send {}
pub(crate) type LocalStream = Box<dyn LocalIo>;
#[cfg(unix)]
#[derive(Debug)]
pub(crate) struct LocalListener(tokio::net::UnixListener);
#[cfg(unix)]
pub(crate) async fn bind(paths: &GethPaths) -> io::Result<LocalListener> {
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<LocalStream> {
Ok(Box::new(
tokio::net::UnixStream::connect(paths.socket_path()).await?,
))
}
#[cfg(unix)]
impl LocalListener {
pub(crate) async fn accept(&mut self) -> io::Result<LocalStream> {
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<tokio::net::windows::named_pipe::NamedPipeServer>,
}
#[cfg(windows)]
pub(crate) async fn bind(paths: &GethPaths) -> io::Result<LocalListener> {
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<LocalStream> {
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<LocalStream> {
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
}

View file

@ -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.

View file

@ -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

View file

@ -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.