Enforce a single daemon per geth home

crates/geth-node/src/daemon.rs

@@ -8,11 +8,15 @@ 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;
+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 {
+        paths: paths.clone(),
+    };
     let config = GethConfig::load(&node.paths.config_file())?;
     let iroh_endpoint = start_daemon_iroh_endpoint(&mut node).await?;
     let mut background_tasks = Vec::new();
@@ -28,8 +32,6 @@ 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;
-        remove_control_socket(&paths)?;
-        let listener = UnixListener::bind(paths.socket_path())?;
         tracing::info!(socket = %paths.socket_path().display(), "geth daemon listening");
         serve_local_control(node.clone(), listener).await
     }
@@ -40,10 +42,38 @@ pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> {
         let _ = task.await;
     }
     let shutdown_result = shutdown_iroh_runtime(&node, iroh_endpoint).await;
-    let socket_cleanup_result = remove_control_socket(&paths);
     serve_result?;
     shutdown_result?;
-    socket_cleanup_result
+    Ok(())
+}
+
+struct ControlSocketGuard {
+    paths: GethPaths,
+}
+
+impl Drop for ControlSocketGuard {
+    fn drop(&mut self) {
+        if let Err(error) = remove_control_socket(&self.paths) {
+            tracing::warn!(%error, "failed to clean up daemon control socket");
+        }
+    }
+}
+
+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) {
+        Ok(listener) => Ok(listener),
+        Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => {
+            Err(NodeError::DaemonAlreadyRunning(paths.home().to_path_buf()))
+        }
+        Err(error) => Err(error.into()),
+    }
 }
 
 fn remove_control_socket(paths: &GethPaths) -> Result<(), NodeError> {
@@ -307,4 +337,34 @@ mod tests {
                 .is_none()
         );
     }
+
+    #[tokio::test]
+    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());
+        paths.ensure_base_dirs().expect("base dirs");
+        let probe = paths.run_dir().join("probe.sock");
+        match std::os::unix::net::UnixListener::bind(&probe) {
+            Ok(listener) => {
+                drop(listener);
+                std::fs::remove_file(probe).expect("remove socket probe");
+            }
+            Err(error) => {
+                eprintln!("skipping live socket assertion; Unix sockets unavailable: {error}");
+                return;
+            }
+        }
+        std::fs::write(paths.socket_path(), b"stale").expect("stale socket fixture");
+
+        let listener = bind_control_socket(&paths)
+            .await
+            .expect("replace stale socket");
+        let error = bind_control_socket(&paths)
+            .await
+            .expect_err("second daemon must be rejected");
+
+        assert!(matches!(error, NodeError::DaemonAlreadyRunning(_)));
+        drop(listener);
+        remove_control_socket(&paths).expect("cleanup socket");
+    }
 }

crates/geth-node/src/lib.rs

@@ -195,6 +195,8 @@ pub enum NodeError {
     OverlayRuntime(String),
     #[error("unsupported platform: {0}")]
     UnsupportedPlatform(String),
+    #[error("geth daemon is already running for home: {0}")]
+    DaemonAlreadyRunning(PathBuf),
 }
 
 #[derive(Clone, Debug)]
@@ -11141,6 +11143,12 @@ mod tests {
             local_control::node_error_code(&NodeError::IrohEndpointUnavailable),
             "iroh_endpoint_unavailable"
         );
+        assert_eq!(
+            local_control::node_error_code(&NodeError::DaemonAlreadyRunning(PathBuf::from(
+                "/tmp/geth"
+            ))),
+            "daemon_already_running"
+        );
     }
 
     #[derive(Debug, PartialEq, Eq)]

crates/geth-node/src/local_control.rs

@@ -516,6 +516,7 @@ pub(crate) fn node_error_code(error: &NodeError) -> &'static str {
         NodeError::Unauthorized(_) => "unauthorized",
         NodeError::PeerNotFound(_) => "peer_not_found",
         NodeError::IrohEndpointUnavailable => "iroh_endpoint_unavailable",
+        NodeError::DaemonAlreadyRunning(_) => "daemon_already_running",
         NodeError::KvNotFound(_) => "kv_not_found",
         NodeError::DbNotFound(_) => "db_not_found",
         NodeError::DocumentNotFound(_) => "document_not_found",