From 6e2291e9a9dfa23cb15e8bdc1a853efb5dba8ec7 Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Sun, 6 Sep 2026 20:38:12 -0700 Subject: [PATCH 1/2] feat(mcp): connect the daemon to its engine over IPC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The single-instance daemon now reaches its hyperd engine over a local IPC channel — a Unix domain socket on Unix/macOS, a named pipe on Windows — instead of a loopback TCP port. Scope (PR A of the daemon IPC plan): only the daemon's *engine* connection moves. The health/control channel and daemon discovery keep their loopback TCP port (daemon.json still carries a numeric health_port); collapsing discovery is deferred to PR B. The per-start callback "dead man's switch" stays TCP. Clean break: no dual-transport and no TCP fallback for the engine connection. - build_params sets TransportMode::Ipc and, on Unix, creates an owner-only (0700) socket directory under the state dir via the existing state_perms::ensure_owner_only_dir, passed through the domain_socket_directory override so hyperd binds inside an already-locked directory (no client-side chmod race). On Windows the named pipe carries hyperd's default owner-only DACL. - The daemon publishes connection_endpoint() rather than the raw endpoint() string: for a Unix socket the raw descriptor reconstructs a non-connectable "/domain/hyper" path (a tab.domain:// scheme artifact) while hyperd actually binds "/hyper". - Engine::is_running()'s daemon-mode probe now connects over the endpoint's transport (UDS / named pipe / TCP) instead of TCP only. - hyperdb-api: the Windows named-pipe name gains a monotonic per-process suffix so several HyperProcess instances in one process (a daemon restart, or the test harness) never reuse a pipe name and fail to bind. Verified end-to-end: the daemon spawns, publishes a socket-path endpoint, and a client connects and runs queries over it; the crash-restart and idle-timeout integration tests pass over IPC. --- hyperdb-api/CHANGELOG.md | 15 ++++ hyperdb-api/src/process.rs | 62 +++++++++++++- hyperdb-mcp/CHANGELOG.md | 14 +++ hyperdb-mcp/src/daemon/run.rs | 136 +++++++++++++++++++++++++++--- hyperdb-mcp/src/engine.rs | 49 ++++++++--- hyperdb-mcp/tests/daemon_tests.rs | 70 ++++++++++----- 6 files changed, 298 insertions(+), 48 deletions(-) diff --git a/hyperdb-api/CHANGELOG.md b/hyperdb-api/CHANGELOG.md index c66a1d2..5113676 100644 --- a/hyperdb-api/CHANGELOG.md +++ b/hyperdb-api/CHANGELOG.md @@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Fixed + +- **IPC transport: a process that spawns several IPC `HyperProcess` instances + now gives each a distinct endpoint on both platforms.** The Windows named-pipe + name and the default Unix domain-socket *directory* were both keyed only on + `std::process::id()` (`hyper-`), so a second instance in the same process + — the MCP daemon restarting `hyperd`, or a test harness driving several in + turn — reused the endpoint. Sequentially this worked only because `Drop` + removed the prior artifact first; two concurrently-live instances collided and + the second surfaced as a 60-second callback-listener timeout. Both the pipe + name and the default socket directory now carry a monotonic per-process suffix + (`hyper--`), so concurrent IPC instances no longer collide. A + caller-supplied `domain_socket_directory` is used verbatim and is the caller's + responsibility to keep unique. + ## [1.0.0-rc.3] - 2026-09-07 ### Removed diff --git a/hyperdb-api/src/process.rs b/hyperdb-api/src/process.rs index 5736610..4b06ee1 100644 --- a/hyperdb-api/src/process.rs +++ b/hyperdb-api/src/process.rs @@ -80,6 +80,25 @@ use tracing::info; use crate::error::{Error, Result}; +/// Monotonic disambiguator for per-process IPC endpoint names. +/// +/// A single process can spawn several `HyperProcess` instances — the MCP daemon +/// restarting `hyperd`, or a test harness driving several daemons in turn — and +/// each IPC instance must claim a *distinct* endpoint. Keying the name on +/// `std::process::id()` alone repeats within a process, so an endpoint not yet +/// released by a prior instance would make the next `bind` fail; this counter +/// gives every instance in the process a unique suffix, while the pid keeps +/// names distinct across processes. +/// +/// This applies symmetrically to both transports: the Windows named-pipe name +/// (`hyper--`) and the default Unix domain-socket *directory* +/// (`hyper--`). Both previously used the bare `hyper-` shape, +/// which collided for two concurrently-live IPC instances in one process (the +/// sequential case only worked because `Drop` removed the per-pid artifact +/// first). +#[cfg(any(unix, windows))] +static IPC_INSTANCE_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + /// Specifies which protocols `HyperProcess` should listen on. /// /// # Examples @@ -251,6 +270,21 @@ impl HyperProcess { Self::start_server(&hyperd_path, parameters) } + /// Builds the default Unix domain-socket *directory* for an IPC instance + /// when the caller has not supplied one. + /// + /// The basename is `hyper--`, where `` comes from the + /// process-wide [`IPC_INSTANCE_SEQ`] counter. The suffix is load-bearing: + /// two concurrently-live IPC `HyperProcess` instances in one process must + /// not share a socket path, or the second bind fails and surfaces as a + /// 60 s callback timeout. The `hyper-` prefix is also load-bearing — `Drop` + /// only cleans up directories whose basename `starts_with("hyper-")`. + #[cfg(unix)] + fn default_socket_dir() -> PathBuf { + let seq = IPC_INSTANCE_SEQ.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!("hyper-{}-{}", std::process::id(), seq)) + } + /// Resolves the hyperd executable from the `HYPERD_PATH` environment /// variable. The value can point at the executable directly, or at a /// directory containing it. @@ -390,8 +424,11 @@ impl HyperProcess { { custom_dir.clone() } else { - // Create a temp directory for the socket - let temp_dir = std::env::temp_dir().join(format!("hyper-{}", std::process::id())); + // Create a temp directory for the socket. The basename carries a + // per-process monotonic suffix (`hyper--`) so two + // concurrently-live IPC instances in one process never share a + // socket path — see `Self::default_socket_dir`. + let temp_dir = Self::default_socket_dir(); std::fs::create_dir_all(&temp_dir).map_err(|e| { Error::connection_with_io("Failed to create socket directory", e) })?; @@ -409,7 +446,8 @@ impl HyperProcess { // Create pipe name for Named Pipes if needed (Windows only) #[cfg(windows)] let pipe_name: Option = if transport_mode == TransportMode::Ipc { - Some(format!("hyper-{}", std::process::id())) + let seq = IPC_INSTANCE_SEQ.fetch_add(1, Ordering::Relaxed); + Some(format!("hyper-{}-{}", std::process::id(), seq)) } else { None }; @@ -1487,6 +1525,24 @@ mod tests { assert_eq!(NO_DEFAULT_PARAMETERS, "no_default_parameters"); } + /// Two IPC instances in one process must derive *distinct* default socket + /// directories, or their sockets collide and the second bind 60 s-timeouts. + /// Also guards the `hyper-` prefix that `Drop`'s cleanup keys on. + #[cfg(unix)] + #[test] + fn default_socket_dir_names_are_unique_and_prefixed() { + let a = HyperProcess::default_socket_dir(); + let b = HyperProcess::default_socket_dir(); + assert_ne!(a, b, "concurrent IPC instances must not share a socket dir"); + for dir in [&a, &b] { + let name = dir.file_name().and_then(|n| n.to_str()).unwrap_or(""); + assert!( + name.starts_with("hyper-"), + "socket dir basename must keep the `hyper-` prefix so Drop cleans it up: {name}" + ); + } + } + #[test] fn test_parameters_with_no_defaults() { let mut params = Parameters::new(); diff --git a/hyperdb-mcp/CHANGELOG.md b/hyperdb-mcp/CHANGELOG.md index 42d8bb0..9358083 100644 --- a/hyperdb-mcp/CHANGELOG.md +++ b/hyperdb-mcp/CHANGELOG.md @@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Changed + +- **The daemon now connects to its `hyperd` engine over a local IPC channel + instead of TCP** — a Unix domain socket on Unix/macOS, a named pipe on + Windows. Only the daemon's engine connection moves: the health/control + channel and daemon discovery keep their existing loopback TCP port, so + `daemon.json` still carries a numeric `health_port`. What changes there is + `hyperd_endpoint` — a daemon-mode session now advertises a socket path (e.g. + `~/.hyperdb/sockets/hyper`) rather than `127.0.0.1:`, and `status` + reports it as `transport: "unix_domain_socket"` (or `"named_pipe"`). The + socket directory is created private to the owner (`0700`) before `hyperd` + binds in it, as ordinary hygiene for a path under the state directory. A + local `--no-daemon` session is unchanged and stays TCP. + ## [1.0.0-rc.3] - 2026-09-07 ### Added diff --git a/hyperdb-mcp/src/daemon/run.rs b/hyperdb-mcp/src/daemon/run.rs index 6d3481f..5db9e55 100644 --- a/hyperdb-mcp/src/daemon/run.rs +++ b/hyperdb-mcp/src/daemon/run.rs @@ -8,6 +8,7 @@ //! `--idle-timeout` flag or `HYPERDB_DAEMON_IDLE_TIMEOUT` env var). When enabled, //! client HEARTBEAT commands reset the idle timer (see [`DaemonState`]). +use std::path::Path; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -94,11 +95,18 @@ pub async fn run_daemon(config: DaemonConfig) -> Result<(), Box/domain/hyper` (a `tab.domain://` scheme artifact), whereas hyperd + // actually binds `/hyper`; `connection_endpoint()` carries the path a + // client can connect to. For TCP and Windows named pipes the two agree. let endpoint = hyper - .endpoint() - .ok_or("hyperd did not report an endpoint")? + .connection_endpoint() + .ok_or("hyperd did not report a connection endpoint")? .to_string(); info!(endpoint = %endpoint, "hyperd started"); @@ -195,15 +203,19 @@ pub fn try_record_restart_attempt(history: &mut Vec, now: Instant) -> R } /// Build the Parameters used for every hyperd spawn (initial start and restarts). -fn build_params() -> std::io::Result { +/// +/// `state_dir` is the resolved daemon state directory (from +/// [`discovery::state_dir`]). Taking it as an argument rather than resolving it +/// internally keeps this unit-testable without touching the process +/// environment. +fn build_params(state_dir: &Path) -> std::io::Result { // The state directory holds `daemon.json`; `logs/` holds `hyperd`'s own // diagnostic logs, which name the endpoint just as `daemon.json` does. // `hyperd` is a separate process writing under its own umask, so // restricting the directory is what covers those files. Both levels are // restricted here so the daemon's own startup establishes the invariant // instead of it depending on the later discovery-file write. - let state_dir = discovery::state_dir()?; - super::state_perms::ensure_owner_only_dir(&state_dir)?; + super::state_perms::ensure_owner_only_dir(state_dir)?; let log_dir = state_dir.join("logs"); super::state_perms::ensure_owner_only_dir(&log_dir)?; @@ -211,7 +223,52 @@ fn build_params() -> std::io::Result { params.set("log_file_max_count", "2"); params.set("log_file_size_limit", "100M"); params.set("log_dir", log_dir.to_string_lossy().as_ref()); - params.set_transport_mode(TransportMode::Tcp); + + // The daemon reaches its engine over a local IPC channel rather than TCP. + params.set_transport_mode(TransportMode::Ipc); + + // On Unix the socket lives in a directory created private to the owner + // *before* hyperd binds inside it. hyperd binds the socket and never + // widens its directory, and creating the directory `0700` up front leaves + // no window in which a client-side tightening would race that bind. On + // Windows the transport is a named pipe with no directory to place, so + // this block compiles out (`domain_socket_directory` is Unix-only). + #[cfg(unix)] + { + // Landmine for PR B: the basename here must NOT start with `hyper-`. + // `HyperProcess::drop` (`hyperdb-api/src/process.rs`) `remove_dir_all`s + // any caller-supplied socket directory whose basename + // `starts_with("hyper-")`, treating it as a temp dir it owns. `sockets` + // is safe; renaming it to e.g. `hyper-sockets` would make Drop delete + // the daemon's persistent state directory contents. + let socket_dir = state_dir.join("sockets"); + super::state_perms::ensure_owner_only_dir(&socket_dir)?; + + // Fast-fail pre-flight. hyperd binds `/hyper`; if another + // hyperd is *genuinely* bound there the connect succeeds, and without + // this guard hyperd would die with "unable to listen on domain socket: + // domain socket is in use" while `HyperProcess::new` masked it as a + // 60-second "Timeout waiting for Hyper to connect to callback listener". + // Reachable today only when two daemons share one state dir (different + // ports). A refused/missing socket (stale file, dead owner) is fine — + // proceed and let hyperd's pid-liveness staleness check reclaim it. This + // keeps the crash-restart path safe: `try_restart_hyperd` reaps the + // SIGKILLed child (`guard.hyper = None`) before calling `build_params`, + // so here the connect is refused and we proceed to rebind. + let socket_path = socket_dir.join("hyper"); + if std::os::unix::net::UnixStream::connect(&socket_path).is_ok() { + return Err(std::io::Error::new( + std::io::ErrorKind::AddrInUse, + format!( + "another hyperd is already bound at {}", + socket_path.display() + ), + )); + } + + params.set_domain_socket_directory(socket_dir); + } + Ok(params) } @@ -316,15 +373,29 @@ fn try_restart_hyperd( // Drop the old hyperd. For an already-exited process this is near-instant; // for a still-alive process, Drop waits up to ~5s for graceful shutdown. + // + // Load-bearing ordering: this drop MUST precede the respawn below. A + // SIGKILLed hyperd leaves a zombie until its `Child` handle is reaped, and a + // zombie still looks alive to hyperd's stale-socket check (pid-liveness on + // `/hyper.pid`) *and* to `build_params`' pre-flight connect. Dropping + // the handle here reaps the child so the replacement can reclaim the socket; + // reordering the respawn before this line wedges the new hyperd for the full + // 60-second callback timeout. guard.hyper = None; // Spawn the replacement. - let params = build_params().map_err(|e| RestartError::SpawnFailed(e.to_string()))?; + let state_dir = discovery::state_dir().map_err(|e| RestartError::SpawnFailed(e.to_string()))?; + let params = build_params(&state_dir).map_err(|e| RestartError::SpawnFailed(e.to_string()))?; let new_hyper = HyperProcess::new(None, Some(¶ms)) .map_err(|e| RestartError::SpawnFailed(e.to_string()))?; + // See `run_daemon`: publish the connectable `connection_endpoint()`, not + // the raw callback descriptor, so a Unix socket path is the one hyperd + // actually bound. let new_endpoint = new_hyper - .endpoint() - .ok_or_else(|| RestartError::SpawnFailed("hyperd did not report endpoint".into()))? + .connection_endpoint() + .ok_or_else(|| { + RestartError::SpawnFailed("hyperd did not report a connection endpoint".into()) + })? .to_string(); // Publish the new endpoint, discovery file first and STATUS second. @@ -381,3 +452,46 @@ async fn shutdown_signal() { ctrl_c.await.ok(); } } + +#[cfg(all(test, unix))] +mod tests { + use std::os::unix::fs::PermissionsExt as _; + + use tempfile::TempDir; + + use super::*; + + /// The daemon's Unix domain socket lives in a directory it creates private + /// to the owner *before* hyperd binds inside it. This reads the mode + /// straight back off the directory `build_params` created and handed to + /// `HyperProcess` through the `domain_socket_directory` override, so a + /// regression that dropped the `ensure_owner_only_dir` call — leaving the + /// directory at its umask-derived mode — would surface here. + #[test] + fn build_params_creates_owner_only_socket_directory() { + let tmp = TempDir::new().unwrap(); + let state_dir = tmp.path().join("state"); + + let params = build_params(&state_dir).expect("build_params should create the state layout"); + + let socket_dir = params + .domain_socket_directory() + .expect("IPC params must carry a Unix domain socket directory"); + assert_eq!( + socket_dir, + state_dir.join("sockets"), + "the socket directory must be a dedicated subdirectory under the state directory" + ); + + let mode = std::fs::metadata(socket_dir) + .expect("the socket directory must exist on disk") + .permissions() + .mode() + & 0o777; + assert_eq!( + mode, 0o700, + "the daemon's socket directory must be owner-only before hyperd binds in it, \ + got {mode:04o}" + ); + } +} diff --git a/hyperdb-mcp/src/engine.rs b/hyperdb-mcp/src/engine.rs index 3ec370f..000f2c2 100644 --- a/hyperdb-mcp/src/engine.rs +++ b/hyperdb-mcp/src/engine.rs @@ -663,14 +663,17 @@ impl Engine { /// Whether the backing `hyperd` is currently reachable. /// /// In local mode, delegates to the owned `HyperProcess`. In daemon mode, - /// probes the cached libpq `daemon_endpoint` directly with a short-timeout - /// TCP connect — the same endpoint queries run against. This reflects - /// *current* liveness of the resource the engine actually depends on, and - /// is robust to two failure modes the health-port PING is not: + /// probes the cached libpq `daemon_endpoint` directly by connecting over + /// whichever transport it names (a Unix domain socket, a named pipe, or + /// TCP) — the same endpoint queries run against. This reflects *current* + /// liveness of the resource the engine actually depends on, and is robust + /// to two failure modes the health-port PING is not: /// - the health port being unreachable (stale `daemon.json`, /// port-scan-adopted daemon, firewall) while the libpq endpoint serves; - /// - the daemon restarting `hyperd` on a new port, leaving the cached - /// endpoint stale (the probe then correctly reports `false`). + /// - the daemon restarting `hyperd` at a new endpoint, leaving the cached + /// one stale (the probe then correctly reports `false`); the Unix socket + /// path is stable across restarts, so there the same probe instead + /// observes the live replacement. /// /// Falls back to discovery (`daemon.json` + health-port PING) only when no /// endpoint has been cached yet (before the first connection attempt). @@ -2515,17 +2518,39 @@ pub fn describe_endpoint(endpoint: &str) -> Value { }) } -/// Cheap liveness probe for a daemon-mode `hyperd`: attempt a short-timeout -/// TCP connect to `endpoint` (`host:port`). Returns `true` if the connect -/// succeeds (something is listening). A bare connect is sufficient here — we -/// only need to know the port the engine's libpq connection targets is still -/// accepting connections, not to perform a full protocol round-trip. +/// Cheap liveness probe for a daemon-mode `hyperd`: attempt a short connect +/// over whichever transport the `endpoint` names — a Unix domain socket path +/// (`/…`), a Windows named pipe (`\\…`), or a TCP `host:port` — and return +/// `true` if it succeeds (something is listening). A bare connect is sufficient +/// here: we only need to know whether hyperd is reachable right now, not to +/// perform a full protocol round-trip. The TCP branch bounds itself with a +/// short timeout; a local UDS/pipe connect is effectively instant. fn probe_endpoint_alive(endpoint: &str) -> bool { + // The daemon reaches hyperd over IPC, so the cached endpoint is a Unix + // domain socket path on Unix or a named pipe on Windows; only a local + // engine's private hyperd is TCP. Probe over whichever transport the + // endpoint names, mirroring how `Connection::connect` routes it, and drop + // the connection immediately — this answers only "is hyperd reachable + // now?". A local UDS/pipe connect is effectively instant, so unlike the + // TCP branch it needs no timeout. + #[cfg(unix)] + if endpoint.starts_with('/') { + return std::os::unix::net::UnixStream::connect(endpoint).is_ok(); + } + #[cfg(windows)] + if endpoint.starts_with(r"\\") { + return std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(endpoint) + .is_ok(); + } + use std::net::ToSocketAddrs; const PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(300); match endpoint.to_socket_addrs() { // Probe each resolved address, short-circuiting on the first that - // accepts a connection. `daemon_endpoint` is normally a single + // accepts a connection. A TCP `daemon_endpoint` is normally a single // `127.0.0.1:PORT`, so this is one connect in the common case. Ok(mut addrs) => { addrs.any(|addr| std::net::TcpStream::connect_timeout(&addr, PROBE_TIMEOUT).is_ok()) diff --git a/hyperdb-mcp/tests/daemon_tests.rs b/hyperdb-mcp/tests/daemon_tests.rs index 498d9cf..13b7477 100644 --- a/hyperdb-mcp/tests/daemon_tests.rs +++ b/hyperdb-mcp/tests/daemon_tests.rs @@ -1615,12 +1615,11 @@ fn hyperd_monitor_detects_killed_hyperd_and_restarts() { .expect("daemon should restart hyperd within 12s"); // The new endpoint must be reachable. Don't assert it differs from the old — - // port reuse is permitted by the OS. - let probe = std::net::TcpStream::connect_timeout( - &new_endpoint.parse().expect("valid endpoint"), - Duration::from_secs(2), + // the OS may reuse a TCP port, and the Unix socket path is fixed by design. + assert!( + endpoint_accepts_connection(&new_endpoint), + "new hyperd endpoint should be reachable" ); - assert!(probe.is_ok(), "new hyperd endpoint should be reachable"); } #[cfg(unix)] @@ -1648,11 +1647,10 @@ fn client_report_triggers_restart_after_kill() { wait_for_live_hyperd_after_kill(daemon.info.health_port, &daemon.info.hyperd_endpoint, 12) .expect("daemon should restart hyperd within 12s after report"); - let probe = std::net::TcpStream::connect_timeout( - &new_endpoint.parse().expect("valid endpoint"), - Duration::from_secs(2), + assert!( + endpoint_accepts_connection(&new_endpoint), + "new hyperd endpoint should be reachable" ); - assert!(probe.is_ok(), "new hyperd endpoint should be reachable"); } #[cfg(unix)] @@ -2059,19 +2057,31 @@ impl Drop for TestDaemon { } } -/// Locate the `hyperd` process by matching the listen-port portion of an -/// endpoint string like `127.0.0.1:54321` against `lsof`'s view of TCP ports. -/// Returns the PID of whichever process owns the port. Unix-only. +/// Locate the `hyperd` process serving a daemon endpoint via `lsof`. Over IPC +/// the endpoint is a Unix domain socket path (`/sockets/hyper`), so +/// match the process holding that socket file open; for a TCP `host:port` +/// endpoint, match the process listening on the port. Returns the PID of +/// whichever process owns it. Unix-only. #[cfg(unix)] fn find_hyperd_pid_for_endpoint(endpoint: &str) -> Option { use std::process::Command; - let port = endpoint.rsplit(':').next()?.parse::().ok()?; - // `lsof -nP -iTCP: -sTCP:LISTEN -t` prints just the PID(s) listening on that port. - let output = Command::new("lsof") - .args(["-nP", &format!("-iTCP:{port}"), "-sTCP:LISTEN", "-t"]) - .output() - .ok()?; + // `lsof -t` prints just the PID(s). A socket path is passed as a + // positional file argument; a TCP endpoint selects the listening socket by + // port. On the socket-path branch, constrain to command `hyperd` (`-c + // hyperd`) so a client sharing the socket can't be the PID the caller then + // kills — the TCP branch already narrows to the listener via `-sTCP:LISTEN`. + let output = if endpoint.starts_with('/') { + Command::new("lsof") + .args(["-nP", "-t", "-c", "hyperd", endpoint]) + .output() + } else { + let port = endpoint.rsplit(':').next()?.parse::().ok()?; + Command::new("lsof") + .args(["-nP", &format!("-iTCP:{port}"), "-sTCP:LISTEN", "-t"]) + .output() + } + .ok()?; if !output.status.success() { return None; } @@ -2084,6 +2094,21 @@ fn find_hyperd_pid_for_endpoint(endpoint: &str) -> Option { .and_then(|s| s.parse::().ok()) } +/// Whether `endpoint` currently accepts a connection, over whichever transport +/// it names: a Unix domain socket path (`/…`) or a TCP `host:port`. Mirrors how +/// a client would reach it, so the restart tests can gate on real reachability +/// regardless of the daemon's transport. Unix-only. +#[cfg(unix)] +fn endpoint_accepts_connection(endpoint: &str) -> bool { + if endpoint.starts_with('/') { + std::os::unix::net::UnixStream::connect(endpoint).is_ok() + } else if let Ok(addr) = endpoint.parse::() { + std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(500)).is_ok() + } else { + false + } +} + /// Kill the given PID with SIGKILL. Unix-only. #[cfg(unix)] fn kill_pid(pid: u32) { @@ -2123,12 +2148,14 @@ fn wait_for_live_hyperd_after_kill( // Phase 1: the killed hyperd must stop accepting connections. Until it // does, any endpoint we observe could still be its soon-to-close socket. - let killed_addr = killed_endpoint.parse::().ok()?; + // For the Unix socket the path is fixed across restarts, but the daemon's + // monitor only reacts on its ~5s tick, so there is always a downtime window + // in which the path refuses a connection before the replacement binds it. loop { if Instant::now() >= deadline { return None; } - if std::net::TcpStream::connect_timeout(&killed_addr, Duration::from_millis(500)).is_err() { + if !endpoint_accepts_connection(killed_endpoint) { break; } std::thread::sleep(Duration::from_millis(25)); @@ -2140,8 +2167,7 @@ fn wait_for_live_hyperd_after_kill( if let Ok(response) = health::send_command(health_port, "STATUS") && let Ok(parsed) = serde_json::from_str::(response.trim()) && let Some(endpoint) = parsed["hyperd_endpoint"].as_str() - && let Ok(addr) = endpoint.parse::() - && std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(500)).is_ok() + && endpoint_accepts_connection(endpoint) { return Some(endpoint.to_string()); } From 0516aa56b8da5fb84fa109640783405025d382b4 Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Mon, 7 Sep 2026 01:37:36 -0700 Subject: [PATCH 2/2] test(mcp): kill the right hyperd in restart tests under parallel load (#310) find_hyperd_pid_for_endpoint located hyperd over a Unix domain socket with `lsof -nP -t -c hyperd `. lsof ORs separately-stated selectors unless `-a` is given, so `-c hyperd ` matched *every* running hyperd process, not just the one bound to . Under the parallel test binary that OR union made `.find(first)` return an unrelated concurrent test's hyperd; the restart test then SIGKILLed the wrong process, its own hyperd stayed alive, the daemon's liveness monitor correctly never restarted, and wait_for_live_hyperd_after_kill timed out. This reproduced only on the loaded ubuntu-latest runner (two of three restart tests, both attempts) and never in the serial or macOS runs; the TCP path on main was immune because `-iTCP:{port} -sTCP:LISTEN` already names a unique listener. Add `-a` so the command-name and path selectors are ANDed and only the hyperd actually bound to the socket path is returned. Test-only; no production or public-API change. --- hyperdb-mcp/tests/daemon_tests.rs | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/hyperdb-mcp/tests/daemon_tests.rs b/hyperdb-mcp/tests/daemon_tests.rs index 672fc98..3177717 100644 --- a/hyperdb-mcp/tests/daemon_tests.rs +++ b/hyperdb-mcp/tests/daemon_tests.rs @@ -2112,12 +2112,23 @@ fn find_hyperd_pid_for_endpoint(endpoint: &str) -> Option { // `lsof -t` prints just the PID(s). A socket path is passed as a // positional file argument; a TCP endpoint selects the listening socket by - // port. On the socket-path branch, constrain to command `hyperd` (`-c - // hyperd`) so a client sharing the socket can't be the PID the caller then - // kills — the TCP branch already narrows to the listener via `-sTCP:LISTEN`. + // port. + // + // On the socket-path branch, `-a` is load-bearing (#310). lsof combines + // separately-stated selectors with OR, not AND, unless `-a` is given — so + // `-c hyperd ` means "every process named hyperd, OR anything holding + // open", i.e. the PID of *every* concurrent hyperd, not the one at + // this socket. Under the parallel test binary that OR union made + // `.find(first)` pick an unrelated test's hyperd; killing it left this + // test's hyperd alive, so the daemon's monitor never restarted and + // `wait_for_live_hyperd_after_kill` timed out — reproducibly on the loaded + // ubuntu runner, never in the serial/macOS runs. `-a` ANDs the command-name + // and path selectors so only the hyperd actually bound to `` matches. + // (The TCP branch was never affected: `-iTCP:{port} -sTCP:LISTEN` already + // names a unique listener, which is why the tests passed on TCP `main`.) let output = if endpoint.starts_with('/') { Command::new("lsof") - .args(["-nP", "-t", "-c", "hyperd", endpoint]) + .args(["-nP", "-t", "-a", "-c", "hyperd", endpoint]) .output() } else { let port = endpoint.rsplit(':').next()?.parse::().ok()?;