Make local server metadata atomic and concurrency-safe

crates/clickhousectl/src/error.rs

@@ -135,6 +135,62 @@ pub enum Error {
     #[error("Server '{0}' not found")]
     ServerNotFound(String),
 
+    #[error("Could not {operation} at {}: {source}. {remediation}", path.display())]
+    ServerLock {
+        operation: &'static str,
+        path: PathBuf,
+        remediation: &'static str,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error(
+        "Could not read metadata for server '{name}' at .clickhouse/servers/{name}.json: {source}. Check that the file is readable and retry."
+    )]
+    ServerMetadataRead {
+        name: String,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error(
+        "Permission denied accessing metadata for server '{name}' at .clickhouse/servers/{name}.json. Restore access to the file and its parent directory, then retry."
+    )]
+    ServerMetadataPermission {
+        name: String,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error(
+        "Metadata for server '{name}' at .clickhouse/servers/{name}.json is invalid: {source}. Repair or remove the metadata file, then retry."
+    )]
+    ServerMetadataParse {
+        name: String,
+        #[source]
+        source: serde_json::Error,
+    },
+
+    #[error(
+        "Could not update metadata for server '{name}' at .clickhouse/servers/{name}.json: {source}. Check write access to .clickhouse/servers and retry."
+    )]
+    ServerMetadataWrite {
+        name: String,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error(
+        "Could not verify process identity for server '{name}' (PID {pid}) while attempting to {operation}: {source}. Metadata was preserved. Check that process inspection tools are available and that you have permission to inspect the process, then retry."
+    )]
+    ServerProcessInspection {
+        name: String,
+        pid: u32,
+        operation: &'static str,
+        #[source]
+        source: std::io::Error,
+    },
+
     #[error(
         "No server name was provided and multiple non-default ClickHouse servers exist. Pass a name or run `clickhousectl local server stop-all`; use `clickhousectl local server list` to see available servers."
     )]

crates/clickhousectl/src/local/discovery.rs

@@ -4,7 +4,7 @@
 //! and command-line arguments to recover server metadata (project root, name,
 //! ports, version). Used for orphaned server recovery and global server listing.
 
-use std::{collections::HashMap, process::Command};
+use std::{collections::HashMap, io, path::Path, process::Command};
 
 /// A ClickHouse process discovered via OS-level process inspection.
 #[derive(Debug, Clone)]
@@ -17,12 +17,38 @@ pub struct DiscoveredProcess {
     pub version: Option<String>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ManagedProcessIdentity {
+    Managed,
+    NotManaged,
+}
+
+#[derive(Debug)]
+pub struct ProcessInspectionError {
+    pub operation: &'static str,
+    pub source: io::Error,
+}
+
+impl ProcessInspectionError {
+    fn new(operation: &'static str, source: io::Error) -> Self {
+        Self { operation, source }
+    }
+}
+
+fn process_command(standard_path: &'static str, fallback: &'static str) -> Command {
+    Command::new(if Path::new(standard_path).is_file() {
+        standard_path
+    } else {
+        fallback
+    })
+}
+
 /// Find all running ClickHouse processes started by the CLI and parse their metadata.
 ///
 /// Only returns processes whose cwd matches the `.clickhouse/servers/<name>/data/` pattern,
 /// meaning they were started by this CLI. Other ClickHouse processes are ignored.
 pub fn discover_clickhouse_processes() -> Vec<DiscoveredProcess> {
-    let pids = find_clickhouse_pids();
+    let pids = find_clickhouse_pids().unwrap_or_default();
     let cwds = get_process_cwds(&pids);
     let mut discovered = Vec::new();
 
@@ -39,19 +65,170 @@ pub fn discover_clickhouse_processes() -> Vec<DiscoveredProcess> {
     discovered
 }
 
-/// Find PIDs of running `clickhouse` processes.
-fn find_clickhouse_pids() -> Vec<u32> {
-    let output = Command::new("pgrep").arg("-x").arg("clickhouse").output();
+/// Determine whether a PID is a ClickHouse process in the managed data
+/// directory for the recorded project and server name.
+pub fn inspect_managed_clickhouse_process(
+    pid: u32,
+    project_root: &str,
+    server_name: &str,
+) -> Result<ManagedProcessIdentity, ProcessInspectionError> {
+    if pid == 0 || i32::try_from(pid).is_err() {
+        return Ok(ManagedProcessIdentity::NotManaged);
+    }
+    if !process_exists(pid).map_err(|source| {
+        ProcessInspectionError::new("check whether the recorded process exists", source)
+    })? {
+        return Ok(ManagedProcessIdentity::NotManaged);
+    }
 
-    match output {
-        Ok(out) if out.status.success() => String::from_utf8_lossy(&out.stdout)
+    inject_inspection_failure("command", "inspect the recorded process executable")?;
+    let is_clickhouse = match (find_clickhouse_pids(), get_process_cmdline(pid)) {
+        (Ok(pids), cmdline) => {
+            pids.contains(&pid) || cmdline.is_ok_and(|cmdline| cmdline_is_clickhouse(&cmdline))
+        }
+        (Err(_), Ok(cmdline)) => cmdline_is_clickhouse(&cmdline),
+        (Err(pgrep_error), Err(ps_error)) => {
+            if !process_exists(pid).map_err(|source| {
+                ProcessInspectionError::new("recheck whether the recorded process exists", source)
+            })? {
+                return Ok(ManagedProcessIdentity::NotManaged);
+            }
+            return Err(ProcessInspectionError::new(
+                "inspect the recorded process executable",
+                io::Error::other(format!(
+                    "pgrep failed: {pgrep_error}; ps failed: {ps_error}"
+                )),
+            ));
+        }
+    };
+    if !is_clickhouse {
+        return Ok(ManagedProcessIdentity::NotManaged);
+    }
+
+    inject_inspection_failure("cwd", "read the recorded process working directory")?;
+    let cwd = match get_process_cwd(pid) {
+        Ok(Some(cwd)) => cwd,
+        Ok(None) => {
+            if !process_exists(pid).map_err(|source| {
+                ProcessInspectionError::new("recheck whether the recorded process exists", source)
+            })? {
+                return Ok(ManagedProcessIdentity::NotManaged);
+            }
+            return Err(ProcessInspectionError::new(
+                "read the recorded process working directory",
+                io::Error::other("process inspection returned no working directory"),
+            ));
+        }
+        Err(source) => {
+            if !process_exists(pid).map_err(|recheck_source| {
+                ProcessInspectionError::new(
+                    "recheck whether the recorded process exists",
+                    recheck_source,
+                )
+            })? {
+                return Ok(ManagedProcessIdentity::NotManaged);
+            }
+            return Err(ProcessInspectionError::new(
+                "read the recorded process working directory",
+                source,
+            ));
+        }
+    };
+    let expected_cwd = Path::new(project_root)
+        .join(".clickhouse")
+        .join("servers")
+        .join(server_name)
+        .join("data");
+
+    inject_inspection_failure("canonicalize", "resolve the managed process paths")?;
+    let actual = Path::new(&cwd).canonicalize().map_err(|source| {
+        ProcessInspectionError::new("resolve the recorded process working directory", source)
+    })?;
+    let expected = expected_cwd.canonicalize().map_err(|source| {
+        ProcessInspectionError::new("resolve the expected managed data directory", source)
+    })?;
+
+    Ok(if actual == expected {
+        ManagedProcessIdentity::Managed
+    } else {
+        ManagedProcessIdentity::NotManaged
+    })
+}
+
+/// Find PIDs of running `clickhouse` processes.
+fn find_clickhouse_pids() -> io::Result<Vec<u32>> {
+    let output = process_command("/usr/bin/pgrep", "pgrep")
+        .arg("-x")
+        .arg("clickhouse")
+        .output()?;
+
+    match output.status.code() {
+        Some(0) => Ok(String::from_utf8_lossy(&output.stdout)
             .lines()
             .filter_map(|line| line.trim().parse::<u32>().ok())
-            .collect(),
-        _ => Vec::new(),
+            .collect()),
+        Some(1) => Ok(Vec::new()),
+        _ => Err(command_error("pgrep", &output)),
     }
 }
 
+fn cmdline_is_clickhouse(cmdline: &str) -> bool {
+    cmdline.split_whitespace().take(2).any(|arg| {
+        Path::new(arg)
+            .file_name()
+            .is_some_and(|name| name == "clickhouse")
+    })
+}
+
+fn process_exists(pid: u32) -> io::Result<bool> {
+    let pid = i32::try_from(pid).map_err(io::Error::other)?;
+    if unsafe { libc::kill(pid, 0) } == 0 {
+        return Ok(true);
+    }
+
+    let source = io::Error::last_os_error();
+    match source.raw_os_error() {
+        Some(libc::ESRCH) => Ok(false),
+        Some(libc::EPERM) => Ok(true),
+        _ => Err(source),
+    }
+}
+
+fn command_error(command: &str, output: &std::process::Output) -> io::Error {
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    io::Error::other(format!(
+        "{command} exited with {}{}",
+        output.status,
+        if stderr.trim().is_empty() {
+            String::new()
+        } else {
+            format!(": {}", stderr.trim())
+        }
+    ))
+}
+
+#[cfg(test)]
+fn inject_inspection_failure(
+    failure: &str,
+    operation: &'static str,
+) -> Result<(), ProcessInspectionError> {
+    if std::env::var("CHCTL_TEST_PROCESS_INSPECTION_FAILURE").as_deref() == Ok(failure) {
+        return Err(ProcessInspectionError::new(
+            operation,
+            io::Error::other(format!("injected {failure} inspection failure")),
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(not(test))]
+fn inject_inspection_failure(
+    _failure: &str,
+    _operation: &'static str,
+) -> Result<(), ProcessInspectionError> {
+    Ok(())
+}
+
 /// Inspect a single process to extract server metadata from its cwd and cmdline.
 fn inspect_process(pid: u32, cwd: &str) -> Option<DiscoveredProcess> {
     let (project_root, server_name) = parse_server_cwd(cwd)?;
@@ -88,7 +265,7 @@ fn get_process_cwds(pids: &[u32]) -> HashMap<u32, String> {
 
     // -a is required to AND the conditions; without it macOS lsof OR's
     // -d and -p, returning the cwd of every process on the system.
-    let output = Command::new("lsof")
+    let output = process_command("/usr/sbin/lsof", "lsof")
         .args(["-a", "-d", "cwd", "-Fn", "-p", &pid_list])
         .output();
 
@@ -101,6 +278,20 @@ fn get_process_cwds(pids: &[u32]) -> HashMap<u32, String> {
     parse_lsof_cwds(&String::from_utf8_lossy(&output.stdout))
 }
 
+#[cfg(target_os = "macos")]
+fn get_process_cwd(pid: u32) -> io::Result<Option<String>> {
+    let pid_arg = pid.to_string();
+    let output = process_command("/usr/sbin/lsof", "lsof")
+        .args(["-a", "-d", "cwd", "-Fn", "-p", &pid_arg])
+        .output()?;
+    let cwd = parse_lsof_cwds(&String::from_utf8_lossy(&output.stdout)).remove(&pid);
+    if cwd.is_some() || output.status.success() {
+        Ok(cwd)
+    } else {
+        Err(command_error("lsof", &output))
+    }
+}
+
 /// Parse `lsof -Fn` output into process working directories.
 ///
 /// `p<pid>` starts a process record and `n<path>` names its cwd. Other fields
@@ -133,17 +324,27 @@ fn get_process_cwds(pids: &[u32]) -> HashMap<u32, String> {
         .collect()
 }
 
+#[cfg(target_os = "linux")]
+fn get_process_cwd(pid: u32) -> io::Result<Option<String>> {
+    let path = std::fs::read_link(format!("/proc/{pid}/cwd"))?;
+    path.into_os_string().into_string().map(Some).map_err(|_| {
+        io::Error::new(
+            io::ErrorKind::InvalidData,
+            "process working directory is not valid UTF-8",
+        )
+    })
+}
+
 /// Get the command-line string of a process.
-fn get_process_cmdline(pid: u32) -> Option<String> {
-    let output = Command::new("ps")
+fn get_process_cmdline(pid: u32) -> io::Result<String> {
+    let output = process_command("/bin/ps", "ps")
         .args(["-o", "args=", "-p", &pid.to_string()])
-        .output()
-        .ok()?;
+        .output()?;
 
     if output.status.success() {
-        Some(String::from_utf8_lossy(&output.stdout).trim().to_string())
+        Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
     } else {
-        None
+        Err(command_error("ps", &output))
     }
 }
 

crates/clickhousectl/src/local/docker.rs

@@ -921,13 +921,12 @@ pub fn start_existing_blocking(id: &str) -> Result<()> {
 /// Safe to call multiple times in one CLI invocation. When Docker isn't
 /// reachable, `connect()` fails fast (no socket → immediate error, no I/O
 /// timeout) and we return silently.
-pub fn recover_project_postgres_blocking(project_cwd: &str) {
+pub fn recover_project_postgres_blocking(project_cwd: &str) -> Result<()> {
     use crate::local::server::{
-        Engine, ServerInfo, ensure_pg_data_dir, pg_instance_key, save_server_info,
-        server_meta_path_for_recovery,
+        Engine, ServerInfo, pg_instance_key, save_recovered_postgres_server_info,
     };
     let cwd_owned = project_cwd.to_string();
-    let _ = block_on(async move {
+    block_on(async move {
         let docker = match connect().await {
             Ok(d) => d,
             Err(_) => return Ok::<(), Error>(()),
@@ -938,10 +937,6 @@ pub fn recover_project_postgres_blocking(project_cwd: &str) {
         };
         for c in containers {
             let key = pg_instance_key(&c.user_name, &c.major);
-            if server_meta_path_for_recovery(&key).exists() {
-                continue;
-            }
-            let _ = ensure_pg_data_dir(&c.user_name, &c.major);
             let info = ServerInfo {
                 name: key,
                 pid: 0,
@@ -953,10 +948,10 @@ pub fn recover_project_postgres_blocking(project_cwd: &str) {
                 engine: Engine::Postgres,
                 container_id: Some(c.container_id.clone()),
             };
-            let _ = save_server_info(&info);
+            save_recovered_postgres_server_info(&info, &c.user_name, &c.major)?;
         }
         Ok(())
-    });
+    })
 }
 
 #[cfg(test)]