From 178e40edf84cdf43f81adfc15303c94ef1ef4f90 Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Thu, 21 May 2026 00:47:32 -0700 Subject: [PATCH 1/3] feat: single-instance daemon for shared hyperd across MCP clients A lightweight daemon manages one shared hyperd process per user so multiple AI clients (Claude Code, Cursor, VS Code, etc.) can access the same persistent databases simultaneously with reduced resource overhead. Architecture: - TCP port binding as cross-platform single-instance lock - Discovery file at ~/.hyperdb/daemon.json (overridable via HYPERDB_STATE_DIR) - Health protocol (PING/HEARTBEAT/STOP/STATUS) for liveness and idle tracking - Auto-spawn: MCP clients transparently start the daemon if none is running - Idle timeout (default 30 min) with heartbeat-based keep-alive - Ephemeral databases DETACH + delete on session end (Windows-safe) - --no-daemon flag to opt out and use legacy per-client hyperd New files: - hyperdb-mcp/src/daemon/{mod,discovery,health,run,spawn}.rs - hyperdb-mcp/tests/daemon_tests.rs (26 tests: unit + integration) --- Cargo.lock | 15 +- hyperdb-mcp/Cargo.toml | 5 +- hyperdb-mcp/src/daemon/discovery.rs | 125 ++++++ hyperdb-mcp/src/daemon/health.rs | 186 +++++++++ hyperdb-mcp/src/daemon/mod.rs | 21 + hyperdb-mcp/src/daemon/run.rs | 152 +++++++ hyperdb-mcp/src/daemon/spawn.rs | 104 +++++ hyperdb-mcp/src/engine.rs | 170 ++++++-- hyperdb-mcp/src/lib.rs | 1 + hyperdb-mcp/src/main.rs | 133 +++++- hyperdb-mcp/src/server.rs | 54 ++- hyperdb-mcp/tests/daemon_tests.rs | 612 ++++++++++++++++++++++++++++ 12 files changed, 1521 insertions(+), 57 deletions(-) create mode 100644 hyperdb-mcp/src/daemon/discovery.rs create mode 100644 hyperdb-mcp/src/daemon/health.rs create mode 100644 hyperdb-mcp/src/daemon/mod.rs create mode 100644 hyperdb-mcp/src/daemon/run.rs create mode 100644 hyperdb-mcp/src/daemon/spawn.rs create mode 100644 hyperdb-mcp/tests/daemon_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 733d888b..15c8a7d1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1811,7 +1811,7 @@ dependencies = [ [[package]] name = "hyperdb-api" -version = "0.1.2" +version = "0.1.3" dependencies = [ "arrow", "bytes", @@ -1832,7 +1832,7 @@ dependencies = [ [[package]] name = "hyperdb-api-core" -version = "0.1.2" +version = "0.1.3" dependencies = [ "arrow", "base64", @@ -1872,7 +1872,7 @@ dependencies = [ [[package]] name = "hyperdb-api-node" -version = "0.1.2" +version = "0.1.3" dependencies = [ "hyperdb-api", "napi", @@ -1884,7 +1884,7 @@ dependencies = [ [[package]] name = "hyperdb-api-salesforce" -version = "0.1.2" +version = "0.1.3" dependencies = [ "arrow", "base64", @@ -1905,7 +1905,7 @@ dependencies = [ [[package]] name = "hyperdb-bootstrap" -version = "0.1.2" +version = "0.1.3" dependencies = [ "anyhow", "clap", @@ -1923,7 +1923,7 @@ dependencies = [ [[package]] name = "hyperdb-mcp" -version = "0.1.2" +version = "0.1.3" dependencies = [ "arrow", "base64", @@ -1931,6 +1931,7 @@ dependencies = [ "clap", "csv", "hyperdb-api", + "libc", "notify", "parquet", "plotters", @@ -3767,7 +3768,7 @@ dependencies = [ [[package]] name = "sea-query-hyperdb" -version = "0.1.2" +version = "0.1.3" dependencies = [ "sea-query", ] diff --git a/hyperdb-mcp/Cargo.toml b/hyperdb-mcp/Cargo.toml index 17ac7f2c..ae36e8fe 100644 --- a/hyperdb-mcp/Cargo.toml +++ b/hyperdb-mcp/Cargo.toml @@ -22,7 +22,7 @@ path = "src/main.rs" [dependencies] hyperdb-api = { path = "../hyperdb-api", version = "0.1.1" } rmcp = { version = "1.7", features = ["server", "transport-io"] } -tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-std", "signal"] } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-std", "signal", "time"] } serde = { workspace = true } serde_json = { workspace = true, features = ["preserve_order"] } clap = { version = "4", features = ["derive"] } @@ -42,6 +42,9 @@ tokio-util = { version = "0.7", features = ["rt"] } tempfile = { workspace = true } sqlformat = "0.5.0" +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [lints] workspace = true diff --git a/hyperdb-mcp/src/daemon/discovery.rs b/hyperdb-mcp/src/daemon/discovery.rs new file mode 100644 index 00000000..002ffb92 --- /dev/null +++ b/hyperdb-mcp/src/daemon/discovery.rs @@ -0,0 +1,125 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Discovery file management for the single-instance daemon. +//! +//! The daemon writes a JSON file to `~/.hyperdb/daemon.json` containing its +//! PID and the `hyperd` endpoint. Clients read this file to locate the running +//! daemon, validating liveness via a TCP health check before trusting it. + +use std::io; +use std::net::TcpStream; +use std::path::PathBuf; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +use super::DEFAULT_DAEMON_PORT; + +/// Information written by the daemon so clients can discover and connect. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DaemonInfo { + /// OS process ID of the daemon. + pub pid: u32, + /// The `hyperd` libpq endpoint clients should connect to (e.g. `127.0.0.1:54321`). + pub hyperd_endpoint: String, + /// The TCP port the daemon's health listener is bound to. + pub health_port: u16, + /// ISO-8601 timestamp when the daemon started. + pub started_at: String, + /// Version of the daemon binary. + pub version: String, +} + +/// Returns the directory used for daemon state files. +/// +/// Resolution order: +/// 1. `HYPERDB_STATE_DIR` environment variable (if set) +/// 2. `~/.hyperdb/` (where `~` is `HOME` on Unix, `USERPROFILE` on Windows) +/// +/// # Errors +/// Returns an error if neither the env var nor the home directory can be determined. +pub fn state_dir() -> io::Result { + if let Some(dir) = std::env::var_os("HYPERDB_STATE_DIR") { + return Ok(PathBuf::from(dir)); + } + let home = home_dir().ok_or_else(|| { + io::Error::new(io::ErrorKind::NotFound, "cannot determine home directory") + })?; + Ok(home.join(".hyperdb")) +} + +/// Returns the path to the discovery file. +/// +/// # Errors +/// Returns an error if the home directory cannot be determined. +pub fn discovery_file_path() -> io::Result { + Ok(state_dir()?.join("daemon.json")) +} + +/// Write the discovery file atomically (write-to-temp then rename). +/// +/// # Errors +/// Returns an error if the state directory cannot be created or the file cannot be written. +pub fn write_discovery_file(info: &DaemonInfo) -> io::Result<()> { + let dir = state_dir()?; + std::fs::create_dir_all(&dir)?; + + let path = dir.join("daemon.json"); + let tmp_path = dir.join("daemon.json.tmp"); + let json = serde_json::to_string_pretty(info).map_err(|e| io::Error::other(e.to_string()))?; + std::fs::write(&tmp_path, json.as_bytes())?; + // On Windows, rename fails if target exists. Remove stale target first. + let _ = std::fs::remove_file(&path); + std::fs::rename(&tmp_path, &path)?; + Ok(()) +} + +/// Read the discovery file and validate that the daemon is still alive. +/// Returns `None` if no daemon is running (file missing, stale, or unreachable). +pub fn discover() -> Option { + let path = discovery_file_path().ok()?; + let contents = std::fs::read_to_string(&path).ok()?; + let info: DaemonInfo = serde_json::from_str(&contents).ok()?; + + // Validate liveness by connecting to the health port + if is_daemon_alive(info.health_port) { + Some(info) + } else { + // Stale file — daemon crashed. Clean up. + let _ = std::fs::remove_file(&path); + None + } +} + +/// Remove the discovery file (called during graceful shutdown). +pub fn remove_discovery_file() { + if let Ok(path) = discovery_file_path() { + let _ = std::fs::remove_file(&path); + } +} + +/// Check if the daemon is alive by attempting a TCP connection to its health port. +fn is_daemon_alive(port: u16) -> bool { + TcpStream::connect_timeout( + &std::net::SocketAddr::from(([127, 0, 0, 1], port)), + Duration::from_secs(2), + ) + .is_ok() +} + +/// Resolve the daemon health port from environment or default. +pub fn resolve_port() -> u16 { + std::env::var(super::ENV_DAEMON_PORT) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(DEFAULT_DAEMON_PORT) +} + +/// Cross-platform home directory resolution. +fn home_dir() -> Option { + // Try HOME (Unix) then USERPROFILE (Windows) + std::env::var_os("HOME") + .or_else(|| std::env::var_os("USERPROFILE")) + .map(PathBuf::from) +} diff --git a/hyperdb-mcp/src/daemon/health.rs b/hyperdb-mcp/src/daemon/health.rs new file mode 100644 index 00000000..d9a4c695 --- /dev/null +++ b/hyperdb-mcp/src/daemon/health.rs @@ -0,0 +1,186 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! TCP health listener for the daemon. +//! +//! The health listener serves two purposes: +//! 1. **Single-instance lock** — binding the port guarantees at most one daemon per user. +//! 2. **Liveness probe + heartbeat** — clients connect and send simple text commands. +//! +//! Protocol (line-based, newline-terminated): +//! - `PING\n` → `PONG\n` (liveness check) +//! - `HEARTBEAT\n` → `OK\n` (resets idle timer) +//! - `STOP\n` → `STOPPING\n` (triggers graceful shutdown) +//! - `STATUS\n` → JSON line with daemon info + +use std::io::{BufRead, BufReader, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Instant; + +use tracing::{debug, warn}; + +use super::discovery::DaemonInfo; + +/// Handle to the health listener, used to check binding success and manage lifecycle. +#[derive(Debug)] +pub struct HealthListener { + listener: TcpListener, + pub port: u16, +} + +/// Shared state between the health listener and the daemon main loop. +#[derive(Debug)] +pub struct DaemonState { + /// Last time any client sent a heartbeat or query. + pub last_activity: std::sync::Mutex, + /// Signal to shut down the daemon. + pub shutdown: AtomicBool, +} + +impl Default for DaemonState { + fn default() -> Self { + Self::new() + } +} + +impl DaemonState { + pub fn new() -> Self { + Self { + last_activity: std::sync::Mutex::new(Instant::now()), + shutdown: AtomicBool::new(false), + } + } + + /// Record activity (resets idle timer). + /// + /// # Panics + /// Panics if the internal mutex is poisoned. + pub fn touch(&self) { + *self.last_activity.lock().expect("mutex poisoned") = Instant::now(); + } + + /// Duration since the last activity. + /// + /// # Panics + /// Panics if the internal mutex is poisoned. + pub fn idle_duration(&self) -> std::time::Duration { + self.last_activity.lock().expect("mutex poisoned").elapsed() + } + + pub fn request_shutdown(&self) { + self.shutdown.store(true, Ordering::Release); + } + + pub fn should_shutdown(&self) -> bool { + self.shutdown.load(Ordering::Acquire) + } +} + +impl HealthListener { + /// Try to bind the health port. + /// + /// # Errors + /// Returns `Err` if the port is already in use (another daemon is running) + /// or the bind fails for another reason. + pub fn bind(port: u16) -> std::io::Result { + let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port)); + let listener = TcpListener::bind(addr)?; + listener.set_nonblocking(true)?; + let port = listener.local_addr()?.port(); + Ok(Self { listener, port }) + } + + /// Run the health listener loop. Spawns per-connection threads until shutdown. + /// Consumes `self` because this is intended to be called from a dedicated thread. + #[expect( + clippy::needless_pass_by_value, + reason = "Arc and DaemonInfo are cloned into per-connection threads" + )] + pub fn run(self, state: Arc, info: DaemonInfo) { + loop { + if state.should_shutdown() { + break; + } + + match self.listener.accept() { + Ok((stream, _addr)) => { + let state = Arc::clone(&state); + let info = info.clone(); + std::thread::spawn(move || { + handle_client(stream, &state, &info); + }); + } + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(std::time::Duration::from_millis(100)); + } + Err(e) => { + warn!(error = %e, "health listener accept error"); + std::thread::sleep(std::time::Duration::from_millis(500)); + } + } + } + debug!("health listener shut down"); + } +} + +#[expect( + clippy::needless_pass_by_value, + reason = "TcpStream must be owned for BufReader" +)] +fn handle_client(stream: TcpStream, state: &DaemonState, info: &DaemonInfo) { + let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(5))); + let mut reader = BufReader::new(&stream); + let mut writer = &stream; + let mut line = String::new(); + + loop { + line.clear(); + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => { + let cmd = line.trim(); + let response = match cmd { + "PING" => "PONG\n".to_string(), + "HEARTBEAT" => { + state.touch(); + "OK\n".to_string() + } + "STOP" => { + state.request_shutdown(); + "STOPPING\n".to_string() + } + "STATUS" => { + let json = serde_json::to_string(info).unwrap_or_default(); + format!("{json}\n") + } + _ => "ERR unknown command\n".to_string(), + }; + if writer.write_all(response.as_bytes()).is_err() { + break; + } + } + Err(_) => break, + } + } +} + +/// Send a command to the daemon's health port and return the response. +/// +/// # Errors +/// Returns an error if the connection fails or the response cannot be read. +pub fn send_command(port: u16, command: &str) -> std::io::Result { + let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port)); + let mut stream = TcpStream::connect_timeout(&addr, std::time::Duration::from_secs(2))?; + stream.set_read_timeout(Some(std::time::Duration::from_secs(5)))?; + + let msg = format!("{command}\n"); + stream.write_all(msg.as_bytes())?; + stream.flush()?; + + let mut reader = BufReader::new(&stream); + let mut response = String::new(); + reader.read_line(&mut response)?; + Ok(response) +} diff --git a/hyperdb-mcp/src/daemon/mod.rs b/hyperdb-mcp/src/daemon/mod.rs new file mode 100644 index 00000000..8f870a70 --- /dev/null +++ b/hyperdb-mcp/src/daemon/mod.rs @@ -0,0 +1,21 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Single-instance daemon for sharing a `hyperd` process across MCP clients. + +pub mod discovery; +pub mod health; +pub mod run; +pub mod spawn; + +/// Default TCP port the daemon binds for health checks and single-instance locking. +pub const DEFAULT_DAEMON_PORT: u16 = 7484; + +/// Default idle timeout in seconds before the daemon shuts down. +pub const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 30 * 60; // 30 minutes + +/// Environment variable to override the daemon port. +pub const ENV_DAEMON_PORT: &str = "HYPERDB_DAEMON_PORT"; + +/// Environment variable to override the idle timeout (seconds). +pub const ENV_IDLE_TIMEOUT: &str = "HYPERDB_DAEMON_IDLE_TIMEOUT"; diff --git a/hyperdb-mcp/src/daemon/run.rs b/hyperdb-mcp/src/daemon/run.rs new file mode 100644 index 00000000..86b95a32 --- /dev/null +++ b/hyperdb-mcp/src/daemon/run.rs @@ -0,0 +1,152 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Daemon main loop: spawns `hyperd`, runs health listener, monitors idle timeout. + +use std::sync::Arc; +use std::time::Duration; + +use tokio::signal; +use tracing::info; + +use hyperdb_api::{HyperProcess, Parameters, TransportMode}; + +use super::discovery::{self, DaemonInfo}; +use super::health::{DaemonState, HealthListener}; +use super::{DEFAULT_IDLE_TIMEOUT_SECS, ENV_IDLE_TIMEOUT}; + +/// Configuration for the daemon process. +#[derive(Debug)] +pub struct DaemonConfig { + pub port: u16, + pub idle_timeout: Duration, +} + +impl DaemonConfig { + pub fn from_args(port: u16, idle_timeout_secs: Option) -> Self { + let idle_timeout_secs = idle_timeout_secs + .or_else(|| { + std::env::var(ENV_IDLE_TIMEOUT) + .ok() + .and_then(|v| v.parse().ok()) + }) + .unwrap_or(DEFAULT_IDLE_TIMEOUT_SECS); + + Self { + port, + idle_timeout: Duration::from_secs(idle_timeout_secs), + } + } +} + +/// Run the daemon. This function blocks until shutdown is triggered. +/// +/// # Errors +/// Returns an error if the health port cannot be bound, `hyperd` fails to start, +/// or the discovery file cannot be written. +pub async fn run_daemon(config: DaemonConfig) -> Result<(), Box> { + // Step 1: Bind health port (single-instance lock) + let listener = HealthListener::bind(config.port).map_err(|e| { + if e.kind() == std::io::ErrorKind::AddrInUse { + format!( + "Another hyperdb daemon is already running on port {}. \ + Use `hyperdb-mcp daemon status` to check or `hyperdb-mcp daemon stop` to stop it.", + config.port + ) + } else { + format!("Failed to bind health port {}: {e}", config.port) + } + })?; + let bound_port = listener.port; + info!(port = bound_port, "daemon health listener bound"); + + // Step 2: Spawn HyperProcess with TCP transport (shared across clients) + let log_dir = discovery::state_dir()?.join("logs"); + std::fs::create_dir_all(&log_dir)?; + + let mut params = Parameters::new(); + 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); + + let hyper = HyperProcess::new(None, Some(¶ms))?; + let endpoint = hyper + .endpoint() + .ok_or("hyperd did not report an endpoint")? + .to_string(); + info!(endpoint = %endpoint, "hyperd started"); + + // Step 3: Write discovery file + let info = DaemonInfo { + pid: std::process::id(), + hyperd_endpoint: endpoint.clone(), + health_port: bound_port, + started_at: chrono::Utc::now().to_rfc3339(), + version: env!("CARGO_PKG_VERSION").to_string(), + }; + discovery::write_discovery_file(&info)?; + info!(path = %discovery::discovery_file_path()?.display(), "discovery file written"); + + // Step 4: Start health listener in background thread + let state = Arc::new(DaemonState::new()); + let health_state = Arc::clone(&state); + let health_info = info.clone(); + let health_handle = std::thread::spawn(move || { + listener.run(health_state, health_info); + }); + + // Step 5: Monitor idle timeout + OS signals + let idle_timeout = config.idle_timeout; + let shutdown_state = Arc::clone(&state); + + tokio::select! { + () = async { + loop { + tokio::time::sleep(Duration::from_secs(10)).await; + if shutdown_state.idle_duration() >= idle_timeout { + info!( + idle_secs = idle_timeout.as_secs(), + "idle timeout reached, shutting down" + ); + shutdown_state.request_shutdown(); + break; + } + if shutdown_state.should_shutdown() { + break; + } + } + } => {} + () = shutdown_signal() => { + info!("received shutdown signal"); + state.request_shutdown(); + } + } + + // Step 6: Graceful shutdown + info!("shutting down daemon"); + discovery::remove_discovery_file(); + drop(hyper); // closes callback connection → hyperd exits + let _ = health_handle.join(); + + Ok(()) +} + +async fn shutdown_signal() { + let ctrl_c = signal::ctrl_c(); + + #[cfg(unix)] + { + let mut sigterm = + signal::unix::signal(signal::unix::SignalKind::terminate()).expect("sigterm handler"); + tokio::select! { + _ = ctrl_c => {} + _ = sigterm.recv() => {} + } + } + + #[cfg(not(unix))] + { + ctrl_c.await.ok(); + } +} diff --git a/hyperdb-mcp/src/daemon/spawn.rs b/hyperdb-mcp/src/daemon/spawn.rs new file mode 100644 index 00000000..deb3de42 --- /dev/null +++ b/hyperdb-mcp/src/daemon/spawn.rs @@ -0,0 +1,104 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Spawn the daemon as a detached background process. +//! +//! When an MCP client starts and no daemon is running, it spawns one using the +//! current binary with the `daemon` subcommand. The spawned process is fully +//! detached so it outlives the parent MCP session. + +use std::io; +use std::process::Command; +use std::time::{Duration, Instant}; + +use tracing::{debug, info}; + +use super::discovery::{self, DaemonInfo}; + +/// Maximum time to wait for the daemon to write its discovery file after spawning. +const SPAWN_TIMEOUT: Duration = Duration::from_secs(10); + +/// Polling interval while waiting for the discovery file. +const POLL_INTERVAL: Duration = Duration::from_millis(100); + +/// Ensure a daemon is running and return its info. +/// If no daemon is detected, spawn one and wait for it to become ready. +/// +/// # Errors +/// Returns an error if the daemon cannot be spawned or does not become ready +/// within the timeout period. +pub fn ensure_daemon(port: u16) -> io::Result { + // Check if already running + if let Some(info) = discovery::discover() { + debug!(endpoint = %info.hyperd_endpoint, "daemon already running"); + return Ok(info); + } + + info!("no running daemon detected, spawning one"); + spawn_detached(port)?; + wait_for_daemon() +} + +/// Spawn `hyperdb-mcp daemon` as a fully detached background process. +fn spawn_detached(port: u16) -> io::Result<()> { + let exe = std::env::current_exe()?; + let port_str = port.to_string(); + + let mut cmd = Command::new(&exe); + cmd.arg("daemon").arg("--port").arg(&port_str); + + // Detach from parent: redirect stdio to null + cmd.stdin(std::process::Stdio::null()); + cmd.stdout(std::process::Stdio::null()); + cmd.stderr(std::process::Stdio::null()); + + // Platform-specific detach flags + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + // SAFETY: setsid() is async-signal-safe per POSIX. Called in pre_exec + // (between fork and exec) to create a new session so the daemon isn't + // killed when the parent terminal/process exits. + unsafe { + cmd.pre_exec(|| { + libc::setsid(); + Ok(()) + }); + } + } + + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + const CREATE_NO_WINDOW: u32 = 0x0800_0000; + const DETACHED_PROCESS: u32 = 0x0000_0008; + cmd.creation_flags(CREATE_NO_WINDOW | DETACHED_PROCESS); + } + + let child = cmd.spawn()?; + info!(pid = child.id(), "daemon process spawned"); + Ok(()) +} + +/// Poll for the discovery file to appear (daemon is ready). +fn wait_for_daemon() -> io::Result { + let start = Instant::now(); + loop { + if let Some(info) = discovery::discover() { + info!(endpoint = %info.hyperd_endpoint, "daemon is ready"); + return Ok(info); + } + + if start.elapsed() >= SPAWN_TIMEOUT { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + format!( + "daemon did not become ready within {} seconds", + SPAWN_TIMEOUT.as_secs() + ), + )); + } + + std::thread::sleep(POLL_INTERVAL); + } +} diff --git a/hyperdb-mcp/src/engine.rs b/hyperdb-mcp/src/engine.rs index f47cc4e2..b225909b 100644 --- a/hyperdb-mcp/src/engine.rs +++ b/hyperdb-mcp/src/engine.rs @@ -38,23 +38,29 @@ //! optimization could use `spawn_blocking` or an async connection API, but the //! current approach is correct and simple. +use crate::daemon; use crate::error::{ErrorCode, McpError}; use crate::schema::ColumnSchema; use hyperdb_api::{Catalog, Connection, CreateMode, HyperProcess, Parameters, SqlType}; use serde_json::{json, Value}; use std::path::{Path, PathBuf}; -/// Owns a running `HyperProcess` and the single `Connection` to its workspace -/// `.hyper` file. All SQL execution flows through this struct. +/// Owns a connection to `hyperd` and the workspace `.hyper` file. All SQL +/// execution flows through this struct. /// -/// Two workspace modes are supported: -/// - **Persistent** — caller supplies a path; the `.hyper` file survives across -/// sessions so tables can be built up incrementally. -/// - **Ephemeral** — a temp directory is created per process; everything is -/// discarded when the server exits. +/// Two process modes: +/// - **Local** — this engine owns the `HyperProcess` subprocess directly. +/// - **Daemon** — a shared daemon manages `hyperd`; the engine only holds a connection. +/// +/// Two workspace modes: +/// - **Persistent** — caller supplies a path; the `.hyper` file survives across sessions. +/// - **Ephemeral** — a temp directory is created per process; discarded on exit. #[derive(Debug)] pub struct Engine { - hyper: HyperProcess, + /// `None` in daemon mode (the daemon owns the process). + hyper: Option, + /// Stored endpoint for daemon mode (the daemon advertises this). + daemon_endpoint: Option, connection: Connection, workspace_path: PathBuf, log_dir: PathBuf, @@ -62,17 +68,10 @@ pub struct Engine { } impl Engine { - #[expect( - clippy::needless_pass_by_value, - reason = "call-site ergonomics: function consumes logically-owned parameters, refactoring signatures is not worth per-site churn" - )] /// Create a new Engine. If `workspace_path` is Some, use that path (persistent mode). /// If None, use a temp file (ephemeral mode). /// - /// Logs from `hyperd` are written to the directory returned by - /// [`resolve_log_dir`]. The same directory should be used by the MCP - /// binary for its own client-side log so operators can find everything - /// in one place when debugging. + /// Connects to the shared daemon if available, falling back to a local `hyperd`. /// /// # Errors /// @@ -85,6 +84,22 @@ impl Engine { /// reports the `hyperd` executable is missing or unreachable via /// `HYPERD_PATH`. pub fn new(workspace_path: Option) -> Result { + Self::new_with_mode(workspace_path, false) + } + + /// Create an engine that bypasses the shared daemon and spawns a private `hyperd`. + /// + /// # Errors + /// Same as [`Self::new`]. + pub fn new_no_daemon(workspace_path: Option) -> Result { + Self::new_with_mode(workspace_path, true) + } + + #[expect( + clippy::needless_pass_by_value, + reason = "Option is consumed by the workspace path resolution logic" + )] + fn new_with_mode(workspace_path: Option, no_daemon: bool) -> Result { let (path, is_persistent) = if let Some(ref p) = workspace_path { let path = PathBuf::from(shellexpand_tilde(p)); if let Some(parent) = path.parent() { @@ -115,6 +130,14 @@ impl Engine { ) })?; + // Try daemon mode first unless disabled + if !no_daemon { + if let Some(engine) = Self::try_daemon_mode(&path, &log_dir, is_persistent)? { + return Ok(engine); + } + } + + // Fall back to spawning a local HyperProcess let mut params = Parameters::new(); params.set("log_file_max_count", "2"); params.set("log_file_size_limit", "100M"); @@ -135,20 +158,11 @@ impl Engine { McpError::new(ErrorCode::InternalError, format!("Failed to connect: {e}")) })?; - // Ensure the `public` schema exists in the workspace database so that - // `load_file`, `load_data`, and unqualified `CREATE TABLE` statements - // resolve without a "could not resolve the schema (3F000)" error. - connection - .execute_command("CREATE SCHEMA IF NOT EXISTS public") - .map_err(|e| { - McpError::new( - ErrorCode::InternalError, - format!("Failed to bootstrap public schema: {e}"), - ) - })?; + bootstrap_public_schema(&connection)?; Ok(Self { - hyper, + hyper: Some(hyper), + daemon_endpoint: None, connection, workspace_path: path, log_dir, @@ -156,22 +170,80 @@ impl Engine { }) } - /// Whether the `hyperd` child process is still alive. + /// Attempt to connect via the shared daemon. Returns `None` if the daemon + /// cannot be reached (falls back to local mode). + fn try_daemon_mode( + path: &Path, + log_dir: &Path, + is_persistent: bool, + ) -> Result, McpError> { + let port = daemon::discovery::resolve_port(); + let info = match daemon::spawn::ensure_daemon(port) { + Ok(info) => info, + Err(e) => { + tracing::debug!(error = %e, "daemon unavailable, falling back to local mode"); + return Ok(None); + } + }; + + let endpoint = &info.hyperd_endpoint; + let connection = Connection::connect( + endpoint, + &path.to_string_lossy(), + CreateMode::CreateIfNotExists, + ) + .map_err(|e| { + McpError::new( + ErrorCode::InternalError, + format!("Failed to connect to daemon hyperd at {endpoint}: {e}"), + ) + })?; + + bootstrap_public_schema(&connection)?; + + // Send heartbeat so daemon knows we're active + let _ = daemon::health::send_command(info.health_port, "HEARTBEAT"); + + Ok(Some(Self { + hyper: None, + daemon_endpoint: Some(info.hyperd_endpoint), + connection, + workspace_path: path.to_path_buf(), + log_dir: log_dir.to_path_buf(), + is_persistent, + })) + } + + /// Whether the backing `hyperd` process is still alive. + /// In daemon mode, checks the daemon health port. pub fn is_running(&self) -> bool { - self.hyper.is_running() + if let Some(ref hyper) = self.hyper { + hyper.is_running() + } else { + // Daemon mode: check if daemon is still reachable + daemon::discovery::discover().is_some() + } } - /// `host:port` endpoint of the hyperd child process. Used by the + /// `host:port` endpoint of the `hyperd` process. Used by the /// watcher to build additional async connections via `hyperdb_api::pool` /// without touching the primary sync connection this engine holds. /// /// # Errors /// - /// Returns [`ErrorCode::InternalError`] if the underlying - /// [`HyperProcess::require_endpoint`] call fails — typically when - /// `hyperd` has exited or never successfully reported an endpoint. + /// Returns [`ErrorCode::InternalError`] if the endpoint is unavailable. pub fn hyperd_endpoint(&self) -> Result { + if let Some(ref endpoint) = self.daemon_endpoint { + return Ok(endpoint.clone()); + } self.hyper + .as_ref() + .ok_or_else(|| { + McpError::new( + ErrorCode::InternalError, + "no hyperd endpoint available".to_string(), + ) + })? .require_endpoint() .map(std::string::ToString::to_string) .map_err(|e| McpError::new(ErrorCode::InternalError, e.to_string())) @@ -766,7 +838,7 @@ impl Engine { }; Ok(json!({ - "hyperd_running": self.hyper.is_running(), + "hyperd_running": self.is_running(), "workspace_path": self.workspace_path.to_string_lossy(), "workspace_mode": if self.is_persistent { "persistent" } else { "ephemeral" }, "table_count": table_count, @@ -1070,6 +1142,34 @@ fn strip_leading_sql_comments(sql: &str) -> &str { s } +impl Drop for Engine { + fn drop(&mut self) { + // In daemon mode with ephemeral databases, DETACH the workspace from hyperd + // (releases the file handle — critical on Windows) then delete the temp file. + if !self.is_persistent && self.daemon_endpoint.is_some() { + let db_name = self.primary_db_name(); + let detach = format!("DETACH DATABASE \"{db_name}\""); + let _ = self.connection.execute_command(&detach); + // Remove the temp directory containing the ephemeral .hyper file + if let Some(parent) = self.workspace_path.parent() { + let _ = std::fs::remove_dir_all(parent); + } + } + } +} + +fn bootstrap_public_schema(connection: &Connection) -> Result<(), McpError> { + connection + .execute_command("CREATE SCHEMA IF NOT EXISTS public") + .map(|_| ()) + .map_err(|e| { + McpError::new( + ErrorCode::InternalError, + format!("Failed to bootstrap public schema: {e}"), + ) + }) +} + /// Minimal `~/` (and `~\` on Windows) expansion. Resolves the home /// directory via `$HOME` on Unix and `%USERPROFILE%` (falling back to /// `%HOMEDRIVE%%HOMEPATH%`) on Windows. `~username/` is not supported — diff --git a/hyperdb-mcp/src/lib.rs b/hyperdb-mcp/src/lib.rs index 4fba0ba3..2d187ca7 100644 --- a/hyperdb-mcp/src/lib.rs +++ b/hyperdb-mcp/src/lib.rs @@ -38,6 +38,7 @@ pub mod attach; pub mod chart; +pub mod daemon; pub mod engine; pub mod error; pub mod export; diff --git a/hyperdb-mcp/src/main.rs b/hyperdb-mcp/src/main.rs index 43a49665..d83917f4 100644 --- a/hyperdb-mcp/src/main.rs +++ b/hyperdb-mcp/src/main.rs @@ -4,6 +4,7 @@ //! Binary entry point for the `hyperdb-mcp` MCP server. //! //! Starts an MCP server on stdio, optionally backed by a persistent workspace. +//! Can also run in daemon mode to manage a shared `hyperd` process. //! //! # Logging //! @@ -19,7 +20,11 @@ //! [`hyperdb_mcp::engine::resolve_log_dir`]). Check the `status` tool for //! the exact paths. -use clap::Parser; +use clap::{Parser, Subcommand}; +use hyperdb_mcp::daemon; +use hyperdb_mcp::daemon::discovery; +use hyperdb_mcp::daemon::health; +use hyperdb_mcp::daemon::run::DaemonConfig; use hyperdb_mcp::engine::{resolve_log_dir, CLIENT_LOG_FILE_NAME}; use hyperdb_mcp::server::HyperMcpServer; use rmcp::ServiceExt; @@ -37,29 +42,107 @@ const VERSION: &str = concat!(env!("CARGO_PKG_VERSION"), ".r", env!("HYPERDB_GIT about = "MCP server for Hyper database analytics" )] struct Cli { + #[command(subcommand)] + command: Option, + /// Path to the `.hyper` workspace file for persistent mode (omit for ephemeral mode) - #[arg(long)] + #[arg(long, global = true)] workspace: Option, /// Run in read-only mode: disables execute, `load_data`, `load_file`, and export to hyper format - #[arg(long)] + #[arg(long, global = true)] read_only: bool, /// Bare mode: disable MCP-managed auxiliary tables. Skips creating /// `_table_catalog` and forces saved queries into in-memory /// (non-persistent) storage, even with --workspace. - #[arg(long)] + #[arg(long, global = true)] bare: bool, + + /// Disable the shared daemon and spawn a private `hyperd` (legacy behavior) + #[arg(long, global = true)] + no_daemon: bool, +} + +#[derive(Subcommand)] +enum Commands { + /// Run as a background daemon managing a shared hyperd process + Daemon { + #[command(subcommand)] + action: Option, + + /// TCP port for health listener and single-instance lock + #[arg(long, default_value_t = daemon::DEFAULT_DAEMON_PORT)] + port: u16, + + /// Idle timeout in seconds before the daemon shuts down + #[arg(long)] + idle_timeout: Option, + }, +} + +#[derive(Subcommand)] +enum DaemonAction { + /// Stop a running daemon + Stop, + /// Show status of the running daemon + Status, } #[tokio::main] async fn main() -> Result<(), Box> { - // Parse CLI first so the log directory matches whatever workspace the - // user requested. This has to happen before tracing is initialized so - // the file layer points at the right place. let cli = Cli::parse(); - // Compute and create the log dir (same logic Engine will use later). + match cli.command { + Some(Commands::Daemon { + action: Some(DaemonAction::Stop), + port, + .. + }) => { + daemon_stop(port); + Ok(()) + } + Some(Commands::Daemon { + action: Some(DaemonAction::Status), + .. + }) => { + daemon_status(); + Ok(()) + } + Some(Commands::Daemon { + action: None, + port, + idle_timeout, + }) => run_daemon_mode(port, idle_timeout).await, + None => run_mcp_mode(cli).await, + } +} + +async fn run_daemon_mode( + port: u16, + idle_timeout: Option, +) -> Result<(), Box> { + // Daemon logs go to ~/.hyperdb/logs/ + let log_dir = discovery::state_dir()?.join("logs"); + std::fs::create_dir_all(&log_dir)?; + + let file_appender = tracing_appender::rolling::never(&log_dir, "hyperdb-daemon.log"); + let (file_writer, _file_guard) = tracing_appender::non_blocking(file_appender); + + let filter = EnvFilter::try_from_default_env() + .unwrap_or_else(|_| EnvFilter::new("info,hyperdb_mcp=debug")); + + tracing_subscriber::registry() + .with(filter) + .with(fmt::layer().with_writer(std::io::stderr)) + .with(fmt::layer().with_writer(file_writer).with_ansi(false)) + .init(); + + let config = DaemonConfig::from_args(port, idle_timeout); + daemon::run::run_daemon(config).await +} + +async fn run_mcp_mode(cli: Cli) -> Result<(), Box> { let log_dir = resolve_log_dir(cli.workspace.as_deref()); if let Err(e) = std::fs::create_dir_all(&log_dir) { eprintln!( @@ -68,13 +151,9 @@ async fn main() -> Result<(), Box> { ); } - // tracing_appender writes via a background thread; we keep the guard - // alive for the duration of `main` so buffered logs get flushed cleanly. let file_appender = tracing_appender::rolling::never(&log_dir, CLIENT_LOG_FILE_NAME); let (file_writer, _file_guard) = tracing_appender::non_blocking(file_appender); - // Default to `info` when RUST_LOG is unset so the log files are actually - // populated. Users can still override via RUST_LOG=debug,hyperdb_api=trace etc. let filter = EnvFilter::try_from_default_env() .unwrap_or_else(|_| EnvFilter::new("info,hyperdb_mcp=debug")); @@ -89,12 +168,40 @@ async fn main() -> Result<(), Box> { workspace = cli.workspace.as_deref().unwrap_or(""), read_only = cli.read_only, bare = cli.bare, + no_daemon = cli.no_daemon, "hyperdb-mcp starting" ); - let server = HyperMcpServer::new(cli.workspace, cli.read_only, cli.bare); + let server = + HyperMcpServer::with_no_daemon(cli.workspace, cli.read_only, cli.bare, cli.no_daemon); let service = server.serve(rmcp::transport::io::stdio()).await?; service.waiting().await?; Ok(()) } + +fn daemon_stop(port: u16) { + match health::send_command(port, "STOP") { + Ok(response) => { + println!("Daemon responded: {}", response.trim()); + } + Err(e) => { + eprintln!("No daemon running on port {port} (or cannot connect): {e}"); + std::process::exit(1); + } + } +} + +fn daemon_status() { + if let Some(info) = discovery::discover() { + println!("Daemon is running:"); + println!(" PID: {}", info.pid); + println!(" Hyperd endpoint: {}", info.hyperd_endpoint); + println!(" Health port: {}", info.health_port); + println!(" Started: {}", info.started_at); + println!(" Version: {}", info.version); + } else { + eprintln!("No daemon is currently running."); + std::process::exit(1); + } +} diff --git a/hyperdb-mcp/src/server.rs b/hyperdb-mcp/src/server.rs index 38c88737..0159eae7 100644 --- a/hyperdb-mcp/src/server.rs +++ b/hyperdb-mcp/src/server.rs @@ -755,6 +755,10 @@ pub struct HyperMcpServer { /// [`crate::saved_queries::SessionStore`] and the catalog is never /// created or updated. bare: bool, + /// Skip the shared daemon and spawn a private `hyperd` (legacy behavior). + no_daemon: bool, + /// Last time a heartbeat was sent to the daemon (debounced to avoid per-call TCP overhead). + last_heartbeat: std::sync::Mutex, // Under rmcp 1.x the router fields are constructed for downstream // macro-generated dispatch but not read through a direct field access // that the compiler can see. Keep them; the `#[tool_router]` / @@ -797,6 +801,25 @@ impl HyperMcpServer { /// workspace file. Useful when callers want a pristine `.hyper` file /// containing only their own data. pub fn new(workspace_path: Option, read_only: bool, bare: bool) -> Self { + Self::with_options(workspace_path, read_only, bare, false) + } + + /// Create a server instance with explicit daemon control. + pub fn with_no_daemon( + workspace_path: Option, + read_only: bool, + bare: bool, + no_daemon: bool, + ) -> Self { + Self::with_options(workspace_path, read_only, bare, no_daemon) + } + + fn with_options( + workspace_path: Option, + read_only: bool, + bare: bool, + no_daemon: bool, + ) -> Self { // Bare mode forces a SessionStore regardless of workspace so the // `_hyperdb_saved_queries` meta-table is never created. let saved_queries: Arc = if bare { @@ -818,6 +841,8 @@ impl HyperMcpServer { workspace_path, read_only, bare, + no_daemon, + last_heartbeat: std::sync::Mutex::new(std::time::Instant::now()), tool_router: Self::tool_router(), prompt_router: Self::prompt_router(), } @@ -927,7 +952,11 @@ impl HyperMcpServer { bare = self.bare, "initializing hyper engine" ); - let engine = Engine::new(self.workspace_path.clone())?; + let engine = if self.no_daemon { + Engine::new_no_daemon(self.workspace_path.clone())? + } else { + Engine::new(self.workspace_path.clone())? + }; tracing::info!( workspace_path = %engine.workspace_path().display(), log_dir = %engine.log_dir().display(), @@ -1045,6 +1074,11 @@ impl HyperMcpServer { // catalog SQL can see errors classified via the normal error // path. No-op in bare or read-only mode. self.ensure_catalog_ready(engine); + // In daemon mode, send a heartbeat so the daemon knows we're still active. + // Debounced to avoid per-call TCP overhead (only sends if >60s since last). + if !self.no_daemon { + self.maybe_send_heartbeat(); + } let result = f(engine); if let Err(e) = &result { tracing::debug!(code = ?e.code, message = %e.message, "tool call returned error"); @@ -1069,6 +1103,24 @@ impl HyperMcpServer { result } + /// Best-effort heartbeat to keep the daemon alive while this client is active. + /// Debounced: only sends if more than 60 seconds have elapsed since the last heartbeat, + /// avoiding a new TCP connection on every tool call. + fn maybe_send_heartbeat(&self) { + const HEARTBEAT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60); + let should_send = self + .last_heartbeat + .lock() + .is_ok_and(|guard| guard.elapsed() >= HEARTBEAT_INTERVAL); + if should_send { + let port = crate::daemon::discovery::resolve_port(); + let _ = crate::daemon::health::send_command(port, "HEARTBEAT"); + if let Ok(mut guard) = self.last_heartbeat.lock() { + *guard = std::time::Instant::now(); + } + } + } + /// Run a closure that accesses the saved-query store. /// /// Some store variants (notably diff --git a/hyperdb-mcp/tests/daemon_tests.rs b/hyperdb-mcp/tests/daemon_tests.rs new file mode 100644 index 00000000..9af31d82 --- /dev/null +++ b/hyperdb-mcp/tests/daemon_tests.rs @@ -0,0 +1,612 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Tests for the single-instance daemon: discovery file, health protocol, +//! idle timeout, and full lifecycle integration with a real `hyperd`. +//! +//! Many tests mutate process-global environment variables (`HYPERDB_STATE_DIR`, +//! `HYPERDB_DAEMON_PORT`) to isolate their state directories. Because env vars +//! are process-global, these tests MUST run sequentially. We enforce this via a +//! shared mutex — every test that touches env vars acquires `ENV_LOCK` first. + +use std::net::TcpListener; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use hyperdb_mcp::daemon::discovery::{self, DaemonInfo}; +use hyperdb_mcp::daemon::health::{self, DaemonState, HealthListener}; +use tempfile::TempDir; + +/// Process-wide lock for tests that mutate environment variables. +/// Cargo runs tests in the same process by default — this prevents races. +static ENV_LOCK: Mutex<()> = Mutex::new(()); + +// ─── Unit tests: DaemonState (no env vars, safe to run in parallel) ─────────── + +#[test] +fn daemon_state_touch_resets_idle_duration() { + let state = DaemonState::new(); + std::thread::sleep(Duration::from_millis(50)); + assert!(state.idle_duration() >= Duration::from_millis(50)); + + state.touch(); + assert!(state.idle_duration() < Duration::from_millis(30)); +} + +#[test] +fn daemon_state_shutdown_flag() { + let state = DaemonState::new(); + assert!(!state.should_shutdown()); + + state.request_shutdown(); + assert!(state.should_shutdown()); +} + +#[test] +fn daemon_state_default_is_equivalent_to_new() { + let default_state = DaemonState::default(); + assert!(!default_state.should_shutdown()); + assert!(default_state.idle_duration() < Duration::from_millis(100)); +} + +// ─── Unit tests: Health protocol (no env vars, safe to run in parallel) ─────── + +#[test] +fn health_listener_bind_succeeds_on_free_port() { + let listener = HealthListener::bind(0).unwrap(); + assert_ne!(listener.port, 0); +} + +#[test] +fn health_listener_second_bind_same_port_fails() { + let listener = HealthListener::bind(0).unwrap(); + let port = listener.port; + + let result = HealthListener::bind(port); + assert!(result.is_err()); + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::AddrInUse); +} + +#[test] +fn health_protocol_ping_pong() { + let (port, _handle, _state) = start_health_listener(); + + let response = health::send_command(port, "PING").unwrap(); + assert_eq!(response.trim(), "PONG"); +} + +#[test] +fn health_protocol_heartbeat_resets_idle() { + let (port, _handle, state) = start_health_listener(); + + std::thread::sleep(Duration::from_millis(50)); + assert!(state.idle_duration() >= Duration::from_millis(50)); + + let response = health::send_command(port, "HEARTBEAT").unwrap(); + assert_eq!(response.trim(), "OK"); + + assert!(state.idle_duration() < Duration::from_millis(30)); +} + +#[test] +fn health_protocol_stop_triggers_shutdown() { + let (port, handle, state) = start_health_listener(); + + assert!(!state.should_shutdown()); + + let response = health::send_command(port, "STOP").unwrap(); + assert_eq!(response.trim(), "STOPPING"); + + assert!(state.should_shutdown()); + + // Health listener should exit its loop + handle.join().unwrap(); +} + +#[test] +fn health_protocol_status_returns_json() { + let (port, _handle, _state) = start_health_listener(); + + let response = health::send_command(port, "STATUS").unwrap(); + let parsed: serde_json::Value = serde_json::from_str(response.trim()).unwrap(); + assert_eq!(parsed["pid"], 12345); + assert_eq!(parsed["hyperd_endpoint"], "127.0.0.1:54321"); +} + +#[test] +fn health_protocol_unknown_command_returns_error() { + let (port, _handle, _state) = start_health_listener(); + + let response = health::send_command(port, "INVALID").unwrap(); + assert!(response.contains("ERR")); +} + +#[test] +fn health_protocol_multi_command_session() { + let (port, _handle, _state) = start_health_listener(); + + let response1 = health::send_command(port, "PING").unwrap(); + assert_eq!(response1.trim(), "PONG"); + + let response2 = health::send_command(port, "STATUS").unwrap(); + let parsed: serde_json::Value = serde_json::from_str(response2.trim()).unwrap(); + assert_eq!(parsed["health_port"], port); + + let response3 = health::send_command(port, "HEARTBEAT").unwrap(); + assert_eq!(response3.trim(), "OK"); +} + +// ─── Unit tests: idle timeout logic (no env vars) ───────────────────────────── + +#[test] +fn daemon_idle_timeout_shuts_down_daemon() { + let state = Arc::new(DaemonState::new()); + let idle_timeout = Duration::from_secs(2); + + let monitor_state = Arc::clone(&state); + let monitor = std::thread::spawn(move || loop { + std::thread::sleep(Duration::from_millis(100)); + if monitor_state.idle_duration() >= idle_timeout { + monitor_state.request_shutdown(); + break; + } + if monitor_state.should_shutdown() { + break; + } + }); + + let start = Instant::now(); + monitor.join().unwrap(); + let elapsed = start.elapsed(); + + assert!(state.should_shutdown()); + assert!(elapsed >= Duration::from_secs(2)); + assert!(elapsed < Duration::from_secs(4)); +} + +#[test] +fn daemon_heartbeat_prevents_idle_shutdown() { + let state = Arc::new(DaemonState::new()); + let idle_timeout = Duration::from_secs(1); + + let monitor_state = Arc::clone(&state); + let heartbeat_state = Arc::clone(&state); + + let heartbeat = std::thread::spawn(move || { + let start = Instant::now(); + while start.elapsed() < Duration::from_millis(1500) { + heartbeat_state.touch(); + std::thread::sleep(Duration::from_millis(200)); + } + }); + + let monitor = std::thread::spawn(move || loop { + std::thread::sleep(Duration::from_millis(100)); + if monitor_state.idle_duration() >= idle_timeout { + monitor_state.request_shutdown(); + break; + } + if monitor_state.should_shutdown() { + break; + } + }); + + heartbeat.join().unwrap(); + let start = Instant::now(); + monitor.join().unwrap(); + let after_heartbeat_stop = start.elapsed(); + + assert!(state.should_shutdown()); + assert!( + after_heartbeat_stop >= Duration::from_millis(800), + "daemon should have waited for idle timeout after heartbeats stopped" + ); +} + +// ─── Unit tests: Discovery file (require ENV_LOCK) ──────────────────────────── + +#[test] +fn discovery_file_write_and_read() { + let _lock = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let _guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); + + let info = DaemonInfo { + pid: 12345, + hyperd_endpoint: "127.0.0.1:54321".to_string(), + health_port: 7484, + started_at: "2026-05-20T10:30:00Z".to_string(), + version: "0.1.3".to_string(), + }; + + discovery::write_discovery_file(&info).unwrap(); + + let path = tmp.path().join("daemon.json"); + assert!(path.exists()); + + let contents = std::fs::read_to_string(&path).unwrap(); + let read_back: DaemonInfo = serde_json::from_str(&contents).unwrap(); + assert_eq!(read_back.pid, 12345); + assert_eq!(read_back.hyperd_endpoint, "127.0.0.1:54321"); + assert_eq!(read_back.health_port, 7484); + assert_eq!(read_back.version, "0.1.3"); +} + +#[test] +fn discovery_file_overwrite_replaces_content() { + let _lock = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let _guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); + + let info1 = DaemonInfo { + pid: 100, + hyperd_endpoint: "127.0.0.1:1111".to_string(), + health_port: 7484, + started_at: "2026-01-01T00:00:00Z".to_string(), + version: "0.1.0".to_string(), + }; + discovery::write_discovery_file(&info1).unwrap(); + + let info2 = DaemonInfo { + pid: 200, + hyperd_endpoint: "127.0.0.1:2222".to_string(), + health_port: 7485, + started_at: "2026-02-02T00:00:00Z".to_string(), + version: "0.2.0".to_string(), + }; + discovery::write_discovery_file(&info2).unwrap(); + + let path = tmp.path().join("daemon.json"); + let contents = std::fs::read_to_string(&path).unwrap(); + let read_back: DaemonInfo = serde_json::from_str(&contents).unwrap(); + assert_eq!(read_back.pid, 200); + assert_eq!(read_back.hyperd_endpoint, "127.0.0.1:2222"); +} + +#[test] +fn remove_discovery_file_deletes_it() { + let _lock = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let _guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); + + let info = DaemonInfo { + pid: 1, + hyperd_endpoint: "127.0.0.1:1".to_string(), + health_port: 7484, + started_at: "2026-01-01T00:00:00Z".to_string(), + version: "0.0.1".to_string(), + }; + discovery::write_discovery_file(&info).unwrap(); + let path = tmp.path().join("daemon.json"); + assert!(path.exists()); + + discovery::remove_discovery_file(); + assert!(!path.exists()); +} + +#[test] +fn discover_returns_none_when_no_file_exists() { + let _lock = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let _guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); + + assert!(discovery::discover().is_none()); +} + +#[test] +fn discover_returns_none_for_stale_file() { + let _lock = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let _guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); + + let info = DaemonInfo { + pid: 99999, + hyperd_endpoint: "127.0.0.1:1".to_string(), + health_port: 1, + started_at: "2026-01-01T00:00:00Z".to_string(), + version: "0.0.1".to_string(), + }; + discovery::write_discovery_file(&info).unwrap(); + + assert!(discovery::discover().is_none()); + + let path = tmp.path().join("daemon.json"); + assert!(!path.exists()); +} + +#[test] +fn resolve_port_uses_env_var() { + let _lock = ENV_LOCK.lock().unwrap(); + let _guard = EnvGuard::set("HYPERDB_DAEMON_PORT", "9999"); + assert_eq!(discovery::resolve_port(), 9999); +} + +#[test] +fn resolve_port_uses_default_when_env_unset() { + let _lock = ENV_LOCK.lock().unwrap(); + let _guard = EnvGuard::remove("HYPERDB_DAEMON_PORT"); + assert_eq!( + discovery::resolve_port(), + hyperdb_mcp::daemon::DEFAULT_DAEMON_PORT + ); +} + +#[test] +fn discover_finds_live_daemon() { + let _lock = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let _guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); + + let (port, _handle, _state) = start_health_listener(); + + let info = DaemonInfo { + pid: 12345, + hyperd_endpoint: "127.0.0.1:54321".to_string(), + health_port: port, + started_at: "2026-05-20T10:30:00Z".to_string(), + version: "0.1.3".to_string(), + }; + discovery::write_discovery_file(&info).unwrap(); + + let discovered = discovery::discover().expect("should discover live daemon"); + assert_eq!(discovered.pid, 12345); + assert_eq!(discovered.health_port, port); +} + +// ─── Integration tests: full daemon lifecycle with real hyperd ───────────────── + +#[test] +fn daemon_mode_engine_connects_to_shared_hyperd() { + let _lock = ENV_LOCK.lock().unwrap(); + let daemon = TestDaemon::start(); + + let tmp = TempDir::new().unwrap(); + let workspace_path = tmp.path().join("test.hyper"); + + let engine = + hyperdb_mcp::engine::Engine::new(Some(workspace_path.to_str().unwrap().to_string())) + .expect("engine should connect to daemon"); + + assert!(engine.is_running()); + + let endpoint = engine.hyperd_endpoint().unwrap(); + assert_eq!(endpoint, daemon.info.hyperd_endpoint); +} + +#[test] +fn daemon_mode_two_engines_share_same_hyperd() { + let _lock = ENV_LOCK.lock().unwrap(); + let _daemon = TestDaemon::start(); + + let tmp1 = TempDir::new().unwrap(); + let tmp2 = TempDir::new().unwrap(); + let path1 = tmp1.path().join("db1.hyper"); + let path2 = tmp2.path().join("db2.hyper"); + + let engine1 = + hyperdb_mcp::engine::Engine::new(Some(path1.to_str().unwrap().to_string())).unwrap(); + + let engine2 = + hyperdb_mcp::engine::Engine::new(Some(path2.to_str().unwrap().to_string())).unwrap(); + + assert_eq!( + engine1.hyperd_endpoint().unwrap(), + engine2.hyperd_endpoint().unwrap() + ); + + engine1.execute_command("CREATE TABLE foo (x INT)").unwrap(); + engine1 + .execute_command("INSERT INTO foo VALUES (42)") + .unwrap(); + + let tables = engine2.describe_tables().unwrap(); + assert!( + tables.iter().all(|t| t["name"] != "foo"), + "engine2 should not see engine1's table" + ); +} + +#[test] +fn daemon_mode_persistent_database_file_survives_engine_drop() { + let _lock = ENV_LOCK.lock().unwrap(); + let _daemon = TestDaemon::start(); + let tmp = TempDir::new().unwrap(); + let path = tmp.path().join("persistent.hyper"); + let path_str = path.to_str().unwrap().to_string(); + + { + let engine = hyperdb_mcp::engine::Engine::new(Some(path_str.clone())).unwrap(); + engine + .execute_command("CREATE TABLE survive (val TEXT)") + .unwrap(); + engine + .execute_command("INSERT INTO survive VALUES ('hello')") + .unwrap(); + } + + assert!( + path.exists(), + "persistent .hyper file should survive engine drop" + ); +} + +#[test] +fn daemon_mode_persistent_engine_data_is_queryable() { + let _lock = ENV_LOCK.lock().unwrap(); + let daemon = TestDaemon::start(); + let tmp = TempDir::new().unwrap(); + let path = tmp.path().join("queryable.hyper"); + let path_str = path.to_str().unwrap().to_string(); + + let engine = hyperdb_mcp::engine::Engine::new(Some(path_str)).unwrap(); + engine + .execute_command("CREATE TABLE items (id INT, name TEXT)") + .unwrap(); + engine + .execute_command("INSERT INTO items VALUES (1, 'alpha'), (2, 'beta')") + .unwrap(); + + let rows = engine + .execute_query_to_json("SELECT * FROM items ORDER BY id") + .unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0]["name"], "alpha"); + assert_eq!(rows[1]["name"], "beta"); + + let resp = health::send_command(daemon.info.health_port, "PING").unwrap(); + assert_eq!(resp.trim(), "PONG"); +} + +#[test] +fn daemon_mode_ephemeral_database_cleaned_up_on_drop() { + let _lock = ENV_LOCK.lock().unwrap(); + let _daemon = TestDaemon::start(); + + let engine = hyperdb_mcp::engine::Engine::new(None).unwrap(); + let workspace_path = engine.workspace_path().to_path_buf(); + + assert!(workspace_path.exists()); + + engine + .execute_command("CREATE TABLE ephemeral_test (id INT)") + .unwrap(); + + drop(engine); + + assert!( + !workspace_path.exists(), + "ephemeral .hyper file should be deleted after engine drop" + ); +} + +// ─── Test helpers ───────────────────────────────────────────────────────────── + +/// Starts a health listener on a random port and returns the port, join handle, +/// and shared state. Does NOT touch env vars — safe for parallel use. +fn start_health_listener() -> (u16, std::thread::JoinHandle<()>, Arc) { + let listener = HealthListener::bind(0).unwrap(); + let port = listener.port; + let state = Arc::new(DaemonState::new()); + let run_state = Arc::clone(&state); + + let info = DaemonInfo { + pid: 12345, + hyperd_endpoint: "127.0.0.1:54321".to_string(), + health_port: port, + started_at: "2026-05-20T10:30:00Z".to_string(), + version: "0.1.3".to_string(), + }; + + let handle = std::thread::spawn(move || { + listener.run(run_state, info); + }); + + // Give the listener a moment to start accepting + std::thread::sleep(Duration::from_millis(50)); + + (port, handle, state) +} + +/// A real daemon running in a background thread for integration tests. +/// Sets `HYPERDB_STATE_DIR` and `HYPERDB_DAEMON_PORT` to isolated values. +/// Caller MUST hold `ENV_LOCK` before calling `start()`. +struct TestDaemon { + info: DaemonInfo, + _state_dir_guard: EnvGuard, + _port_guard: EnvGuard, +} + +impl TestDaemon { + fn start() -> Self { + let tmp = TempDir::new().unwrap(); + // Leak the TempDir so it persists for the lifetime of the test. + let tmp = Box::leak(Box::new(tmp)); + + let state_dir_guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); + + let port = find_free_port(); + let port_guard = EnvGuard::set("HYPERDB_DAEMON_PORT", &port.to_string()); + + // Start the daemon in a background tokio runtime + let daemon_port = port; + std::thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.block_on(async { + let config = hyperdb_mcp::daemon::run::DaemonConfig { + port: daemon_port, + idle_timeout: Duration::from_secs(300), + }; + let _ = hyperdb_mcp::daemon::run::run_daemon(config).await; + }); + }); + + // Wait for daemon to become ready + let start = Instant::now(); + loop { + if let Some(info) = discovery::discover() { + return Self { + info, + _state_dir_guard: state_dir_guard, + _port_guard: port_guard, + }; + } + assert!( + start.elapsed() <= Duration::from_secs(15), + "TestDaemon did not start within 15 seconds" + ); + std::thread::sleep(Duration::from_millis(100)); + } + } +} + +impl Drop for TestDaemon { + fn drop(&mut self) { + let _ = health::send_command(self.info.health_port, "STOP"); + std::thread::sleep(Duration::from_millis(200)); + } +} + +/// Find a free TCP port by binding to port 0 and reading the assigned port. +fn find_free_port() -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.local_addr().unwrap().port() +} + +/// RAII guard that sets/removes an environment variable and restores it on drop. +struct EnvGuard { + key: String, + previous: Option, +} + +impl EnvGuard { + fn set(key: &str, value: &str) -> Self { + let previous = std::env::var(key).ok(); + // SAFETY: Callers hold ENV_LOCK, ensuring no concurrent env var access. + unsafe { std::env::set_var(key, value) }; + Self { + key: key.to_string(), + previous, + } + } + + fn remove(key: &str) -> Self { + let previous = std::env::var(key).ok(); + // SAFETY: Callers hold ENV_LOCK, ensuring no concurrent env var access. + unsafe { std::env::remove_var(key) }; + Self { + key: key.to_string(), + previous, + } + } +} + +impl Drop for EnvGuard { + fn drop(&mut self) { + match &self.previous { + // SAFETY: Callers hold ENV_LOCK for the lifetime of this guard. + Some(val) => unsafe { std::env::set_var(&self.key, val) }, + // SAFETY: Callers hold ENV_LOCK for the lifetime of this guard. + None => unsafe { std::env::remove_var(&self.key) }, + } + } +} From fbccfa81bf1bde9dae235ceef0e495efadc98ede Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Mon, 25 May 2026 00:20:59 -0700 Subject: [PATCH 2/3] docs: describe shared-daemon operating modes and CLI surface The single-instance daemon shipped in the previous commit but the docs still framed persistent workspaces as "experimental, one session at a time" and listed the shared daemon as future work. - README: new "Operating Modes" section covering the two independent dimensions (engine: shared daemon vs --no-daemon; persistence: ephemeral vs --workspace); CLI Reference expanded with the daemon subcommand and HYPERDB_* env vars. - DEVELOPMENT.md: new "Daemon Mode Internals" section. - ROADMAP.md: drops the "Shared hyperd daemon" entry since it shipped. - CHANGELOG.md: unreleased entry summarizing the feature. --- hyperdb-mcp/CHANGELOG.md | 15 +++++++++ hyperdb-mcp/DEVELOPMENT.md | 16 ++++++++-- hyperdb-mcp/README.md | 65 ++++++++++++++++++++++++++++++++++++-- hyperdb-mcp/ROADMAP.md | 34 +++----------------- 4 files changed, 95 insertions(+), 35 deletions(-) diff --git a/hyperdb-mcp/CHANGELOG.md b/hyperdb-mcp/CHANGELOG.md index 482b2543..d0cb8102 100644 --- a/hyperdb-mcp/CHANGELOG.md +++ b/hyperdb-mcp/CHANGELOG.md @@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- **Single-instance `hyperd` daemon** — by default, all MCP clients now + share one `hyperd` process per user instead of each spawning their own. + Multiple AI clients (Claude Code, Cursor, VS Code Copilot, etc.) can + access the same persistent databases simultaneously with reduced + resource overhead. The daemon auto-spawns on first client connect and + shuts down after 30 minutes idle. Pass `--no-daemon` to opt out. +- New `daemon` subcommand: `hyperdb-mcp daemon status` / `daemon stop`. +- New environment variables: `HYPERDB_STATE_DIR`, `HYPERDB_DAEMON_PORT`, + `HYPERDB_DAEMON_IDLE_TIMEOUT`. +- Ephemeral databases now `DETACH DATABASE` before deletion on session + end — required on Windows where the OS enforces file locks on open + Hyper files. + ## [0.1.1] - 2026-05-13 ### Added diff --git a/hyperdb-mcp/DEVELOPMENT.md b/hyperdb-mcp/DEVELOPMENT.md index db1e506f..d7ac0598 100644 --- a/hyperdb-mcp/DEVELOPMENT.md +++ b/hyperdb-mcp/DEVELOPMENT.md @@ -198,11 +198,23 @@ Derived fields are merged into the serialized JSON so callers get a self-contain ## Workspace Modes Internals -- **Ephemeral** — `Engine::new(None)` creates a temp directory (`$TMPDIR/hyperdb-mcp-/`) with a `workspace.hyper` file. Cleaned up on process exit. +- **Ephemeral** — `Engine::new(None)` creates a temp directory (`$TMPDIR/hyperdb-mcp-/`) with a `workspace.hyper` file. Cleaned up on process exit. In daemon mode, `Engine::drop` issues `DETACH DATABASE` (releasing Hyper's file lock — required on Windows) before `remove_dir_all` deletes the temp directory. - **Persistent** — `Engine::new(Some(path))` uses the caller-supplied path. Parent directories are created automatically. `~` is expanded via `$HOME` (no shell crate dependency). Logs (both `hyperd` server logs and the MCP client log) land in the same directory as the workspace file. The `status` tool reports log paths so operators know where to look. +## Daemon Mode Internals + +`Engine::new` defaults to *daemon mode* — it tries `daemon::spawn::ensure_daemon()` first, which discovers an existing daemon via `~/.hyperdb/daemon.json` (overridable via `HYPERDB_STATE_DIR`) or auto-spawns one as a detached background process. The Engine then connects via TCP (`Connection::connect(endpoint, …)`) without owning any `HyperProcess`. + +Falls back to local mode (per-session `hyperd` via `HyperProcess::new`) when the daemon can't be reached, or always when `--no-daemon` is passed. + +Cross-platform single-instance lock is the daemon's TCP health port — bind succeeds for exactly one process per user. Liveness is validated by the discovery flow before trusting the file: a stale `daemon.json` (daemon crashed) is detected and removed. + +The daemon's main loop tracks idle time via `DaemonState::last_activity`. `HEARTBEAT` commands from active clients reset the timer; clients debounce these to once per 60 seconds in `HyperMcpServer::with_engine`. Idle timeout (default 30 min) triggers graceful shutdown: discovery file removed → `hyperd` dropped → health listener exits. + +See `src/daemon/{mod,discovery,health,run,spawn}.rs` for the full implementation. + --- ## Known Tech Debt / Future Work @@ -220,6 +232,6 @@ Logs (both `hyperd` server logs and the MCP client log) land in the same directo ## Forward-Looking Design Notes -Feature-level ideas that aren't bugs or tech debt (which are tracked above) — shared `hyperd` daemon, cross-database tools, catalog awareness for attached databases, `switch_workspace`, and cross-workspace data-movement fallbacks — now live in [ROADMAP.md](ROADMAP.md). That split keeps this file focused on the current codebase and how to work in it, while ROADMAP.md captures the "not built yet but worth thinking about" material. +Feature-level ideas that aren't bugs or tech debt (which are tracked above) — cross-database tools, catalog awareness for attached databases, `switch_workspace`, and cross-workspace data-movement fallbacks — now live in [ROADMAP.md](ROADMAP.md). That split keeps this file focused on the current codebase and how to work in it, while ROADMAP.md captures the "not built yet but worth thinking about" material. --- diff --git a/hyperdb-mcp/README.md b/hyperdb-mcp/README.md index 89030caa..3a8b2ac2 100644 --- a/hyperdb-mcp/README.md +++ b/hyperdb-mcp/README.md @@ -15,6 +15,7 @@ LLMs are powerful at reasoning but cannot natively crunch millions of rows. This ## Features - **Zero setup** — `HyperProcess` auto-starts the Hyper server +- **Shared `hyperd` daemon** — one Hyper process per user, shared across all MCP clients (Claude Code, Cursor, VS Code, etc.) for reduced memory overhead and concurrent access to the same persistent databases - **Any data in** — JSON, CSV, Parquet, Arrow IPC, Apache Iceberg; schema inferred or exact - **SQL at scale** — thousands to billions of rows - **Data out** — export to CSV, Parquet, Apache Iceberg, Arrow IPC, or `.hyper` (Tableau Desktop-ready) @@ -135,7 +136,8 @@ For a **persistent workspace** (tables survive across sessions), add `"args"`: ```json "args": ["--workspace", "/path/to/my-project.hyper"] ``` -This is still **experimental** and will only work with only one session at a time since the Hyper database is locked by Hyper. Each session is isolated and has its own Hyper instance running. Future work will allow multiple sessions to share the same database but requires work to spin up a shared Hyper instance. + +Multiple MCP clients can point at the **same** persistent workspace simultaneously — they all connect through the shared `hyperd` daemon and use Hyper's MVCC transaction isolation. See [Operating Modes](#operating-modes) below. #### Claude Code / AI Suite @@ -159,6 +161,49 @@ Any tool that supports the MCP stdio transport can use this server. Point it at --- +## Operating Modes + +The server has two independent mode dimensions: **how the Hyper engine is run** and **where the database is stored**. + +### Hyper engine + +| Mode | Flag | Behavior | +|---|---|---| +| **Shared daemon** *(default)* | *(none)* | One `hyperd` process per user, shared across all MCP clients. The first client auto-spawns the daemon; subsequent clients discover and reuse it. Idle for 30 minutes → daemon shuts itself down; the next client spawns a fresh one. | +| **Private hyperd** | `--no-daemon` | Each MCP client spawns its own `hyperd` (legacy behavior, one per session). | + +The shared daemon is the bigger win for users running multiple AI clients (Claude Code + Cursor + VS Code) — they all share one Hyper engine instead of spawning three. + +### Database persistence + +| Mode | Flag | Behavior | +|---|---|---| +| **Ephemeral** *(default)* | *(none)* | A temp `.hyper` file is created per session and deleted on exit (DETACH + delete in daemon mode, Windows-safe). | +| **Persistent** | `--workspace ` | Uses the supplied `.hyper` file; survives across sessions. Multiple clients can point at the same path simultaneously. | + +The two dimensions are orthogonal — any combination works. With the default (shared daemon + ephemeral), every client gets its own scratch database living inside the same shared engine; with `--workspace` added, multiple clients can collaborate on the same persistent dataset. + +### Daemon management + +The daemon is normally invisible — it auto-spawns and idle-times-out on its own. For diagnostics: + +```bash +hyperdb-mcp daemon status # Show running daemon (PID, endpoint, started_at, version) +hyperdb-mcp daemon stop # Gracefully shut down the daemon +hyperdb-mcp daemon # Run as a daemon explicitly (rarely needed) +``` + +State files live at `~/.hyperdb/` by default (override with `HYPERDB_STATE_DIR`). + +### Other behavioral flags + +| Flag | Behavior | +|---|---| +| `--read-only` | Disables `execute`, `load_data`, `load_file`, `watch_directory`, `save_query`, `delete_query`, and Hyper-format export. See [Read-Only Mode](#read-only-mode). | +| `--bare` | Skips MCP-managed auxiliary tables (`_table_catalog`); saved queries are kept in-memory only, even with `--workspace`. | + +--- + ## MCP Tools ### One-Shot Tools @@ -639,15 +684,29 @@ Full reference: [Data Cloud SQL Reference](https://developer.salesforce.com/docs ## CLI Reference ``` -hyperdb-mcp [OPTIONS] +hyperdb-mcp [OPTIONS] [COMMAND] + +Commands: + daemon Run as a background daemon managing a shared hyperd process Options: --workspace Path to the `.hyper` workspace file for persistent mode (omit for ephemeral) --read-only Disable mutating tools (execute, load_data, load_file, save_query, delete_query, watch_directory) --bare Skip MCP-managed auxiliary tables (`_table_catalog`) and force saved queries into in-memory storage, even with --workspace + --no-daemon Disable the shared daemon and spawn a private hyperd (legacy per-session behavior) + +Daemon subcommand: + hyperdb-mcp daemon Start the daemon (usually auto-spawned) + hyperdb-mcp daemon stop Gracefully stop the running daemon + hyperdb-mcp daemon status Show running daemon info + hyperdb-mcp daemon --port Override the health/lock port (default 7484) + hyperdb-mcp daemon --idle-timeout Override idle timeout (default 1800 = 30 min) Environment: - HYPERD_PATH Path to hyperd binary (auto-detected if on PATH) + HYPERD_PATH Path to hyperd binary (auto-detected if on PATH) + HYPERDB_STATE_DIR Override daemon state directory (default ~/.hyperdb/) + HYPERDB_DAEMON_PORT Override daemon health/lock port (default 7484) + HYPERDB_DAEMON_IDLE_TIMEOUT Override daemon idle timeout in seconds (default 1800) ``` --- diff --git a/hyperdb-mcp/ROADMAP.md b/hyperdb-mcp/ROADMAP.md index 76282dd5..eb55de94 100644 --- a/hyperdb-mcp/ROADMAP.md +++ b/hyperdb-mcp/ROADMAP.md @@ -8,33 +8,6 @@ Each section follows a loose template: Motivation → Architecture sketch → Es --- -## Shared `hyperd` daemon for cross-workspace JOINs - -**Motivation.** Today, each MCP server (one per entry in `mcp.json`) spawns its own `hyperd` subprocess via `HyperProcess::new()` in `Engine::new`. Two MCP servers ⇒ two `hyperd` processes ⇒ two fully-isolated databases with no ability to JOIN across them. Users who want to query data that spans two workspaces have to either (a) consolidate everything into one workspace, or (b) manually shuffle data between workspaces via the [export bridge](#raw-fallbacks-for-cross-workspace-data-movement) below. - -**Architecture.** Switch the Engine to support *connecting to* an existing `hyperd` rather than always spawning a new one: - -1. New CLI flag: `--hyperd-url tab.tcp://localhost:PORT`. When set, skip `HyperProcess::new` and build a `Connection` directly against the given address. -2. One long-lived `hyperd` daemon (outside any MCP server), managed via `launchd` on macOS / `systemd` on Linux / `hyperd &` in a scratch shell. -3. Each MCP server in `mcp.json` gets `--hyperd-url` pointing at the shared daemon plus its own `--workspace` (the per-instance `.hyper` file it manages). -4. Inside the shared `hyperd`, each MCP server sees its own workspace as the default database via `ATTACH DATABASE ... AS workspace`. To JOIN across workspaces, the LLM issues an additional `ATTACH` for the *other* workspace and then uses fully-qualified table names (`other.public.tablename`). - -**Estimated size.** ~100–200 LOC: -- `src/engine.rs`: split `Engine::new` into `spawn_hyperd_and_connect` vs. `connect_to_existing`, dispatch on the CLI flag. -- `src/main.rs`: add the `--hyperd-url` flag to the clap parser. -- Reconnection handling: when the shared `hyperd` restarts, detect and re-attach workspaces automatically. -- Health check: `status` should report which mode is active (spawned vs. shared) and the daemon URL. - -**Risks / open questions.** -- Port discovery: the shared daemon needs a stable known port, or a discovery file. -- Lifecycle: who starts / stops the shared daemon? Not the MCP server (it might be one of several clients). -- Observability: with many MCP servers sharing one `hyperd`, per-client logs get mixed. May need to rely on `request-id` / `session-id` tagging already in the hyperd log. -- Memory savings on the shared `hyperd` are the motivation, but only really matter once you have ≥3 workspaces — two isn't a big deal on a dev laptop (~150–250 MB idle per `hyperd`). - -**Verdict.** Not urgent for current usage (two workspaces). Add when a concrete "I need to JOIN across sandbox + persistent data right now" use case shows up. - ---- - ## Cross-database tools First-class tools for attaching additional `.hyper` databases and @@ -108,7 +81,7 @@ keep these workarounds in mind: ## `switch_workspace` mid-session tool -Lower-priority than the shared daemon, but conceptually clean: a `switch_workspace(path)` MCP tool that tears down the current Engine and re-instantiates against a different `.hyper` file without a process restart. Also resets the saved-queries store (back to ephemeral/persistent pick), subscription registry, and active watchers. +Conceptually clean: a `switch_workspace(path)` MCP tool that tears down the current Engine and re-instantiates against a different `.hyper` file without a process restart. Also resets the saved-queries store (back to ephemeral/persistent pick), subscription registry, and active watchers. Useful if you'd rather "flip between N workspaces in one chat" than "have N MCP servers in the sidebar". Feasible as ~100 LOC in `server.rs` but introduces subtleties: what happens to in-flight subscriptions, watcher threads with state, and saved queries whose results reference the old workspace's tables. Probably gated behind a CLI flag so it's opt-in. @@ -132,5 +105,6 @@ tables without issuing raw `pg_catalog` SQL: queries will rerun `pg_catalog` anyway). Also punts: remote kinds (`"tcp"` / `"grpc"`) on `attach_database`. -Those need the shared-daemon + credential-profile infrastructure -described above. +Those need credential-profile infrastructure (auth tokens, key +material) that doesn't exist yet — the shared-daemon work landed +already, but it covers only the local-`hyperd` case. From 760575c8acb9bcc3a1dda53fb3f7574ace1e8ae0 Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Mon, 25 May 2026 01:21:33 -0700 Subject: [PATCH 3/3] feat(daemon): detect and restart crashed hyperd MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds liveness monitoring and automatic restart for the daemon's hyperd process. Previously, if hyperd crashed while the daemon kept running, clients would silently fail to connect to a stale endpoint. Detection: - New HyperProcess::has_exited() uses Child::try_wait() — correct on both Unix and Windows, reaps zombies as a side effect. Replaces the Windows-broken kill -0 / always-true heuristic that is_running used. - Daemon polls every 5 seconds. - Clients fast-path the signal via a new REPORT_HYPERD_ERROR health command when they detect a dead endpoint before the polling tick. Restart: - try_restart_hyperd drops old hyperd, spawns replacement with same parameters, atomically rewrites daemon.json with the new endpoint. - STATUS reads stay non-blocking — endpoint info is shared via Arc>, separate from the HyperProcess mutex. - Rate-limited to 3 restarts per 60 seconds; the 4th attempt triggers daemon shutdown so the user sees the failure clearly. Recovery: - Existing ConnectionLost → drop-engine → re-discover path picks up the new endpoint transparently. No MCP tool-handler changes. Tests: - 11 new tests covering has_exited semantics, restart-history math (off-by-N enforcement), the REPORT_HYPERD_ERROR command, and end-to-end engine recovery after SIGKILL. - Adversarial reviewers caught a Windows is_running bug in the plan phase and a spurious-double-restart race in the code phase; both fixed before this commit. Known limitation: hung-but-alive hyperd (TCP listening, query-stuck) is not detected — operator recovery is `hyperdb-mcp daemon stop`. --- hyperdb-api/src/process.rs | 22 ++ hyperdb-api/tests/process_lifecycle_tests.rs | 60 ++++ hyperdb-mcp/CHANGELOG.md | 10 + hyperdb-mcp/DEVELOPMENT.md | 28 ++ hyperdb-mcp/README.md | 10 + hyperdb-mcp/src/daemon/health.rs | 100 +++++- hyperdb-mcp/src/daemon/run.rs | 259 +++++++++++++--- hyperdb-mcp/src/engine.rs | 4 + hyperdb-mcp/src/server.rs | 6 + hyperdb-mcp/tests/daemon_tests.rs | 301 ++++++++++++++++++- 10 files changed, 740 insertions(+), 60 deletions(-) create mode 100644 hyperdb-api/tests/process_lifecycle_tests.rs diff --git a/hyperdb-api/src/process.rs b/hyperdb-api/src/process.rs index da2c0016..3cdb9494 100644 --- a/hyperdb-api/src/process.rs +++ b/hyperdb-api/src/process.rs @@ -1005,6 +1005,28 @@ impl HyperProcess { } } + /// Returns true if the hyperd child process has exited (or no child exists). + /// + /// Uses [`std::process::Child::try_wait`] under the hood, which is correct + /// on both Unix and Windows. On Unix this also reaps any zombie state as a + /// side effect — a hyperd that has been SIGKILLed but not yet `wait()`ed + /// on by the parent will be observed as exited and cleaned up here. + /// + /// Prefer this over [`Self::is_running`] when the caller owns the + /// `HyperProcess` mutably and needs an authoritative liveness signal. + /// `is_running` uses `kill -0` on Unix (which incorrectly reports zombies + /// as alive) and is a no-op on Windows. + pub fn has_exited(&mut self) -> bool { + match self.child.as_mut() { + Some(child) => match child.try_wait() { + Ok(Some(_status)) => true, + Ok(None) => false, + Err(_) => true, + }, + None => true, + } + } + /// Shuts down the Hyper server gracefully with a timeout. /// /// This closes the callback connection, which signals Hyper to shut down gracefully. diff --git a/hyperdb-api/tests/process_lifecycle_tests.rs b/hyperdb-api/tests/process_lifecycle_tests.rs new file mode 100644 index 00000000..fcaab8ca --- /dev/null +++ b/hyperdb-api/tests/process_lifecycle_tests.rs @@ -0,0 +1,60 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Tests for `HyperProcess` lifecycle helpers — specifically the `has_exited` +//! probe used by the hyperdb-mcp daemon's restart logic. + +mod common; + +use common::test_hyper_params; +use hyperdb_api::HyperProcess; + +/// A freshly-spawned hyperd should not appear exited. +#[test] +fn has_exited_returns_false_for_running_hyperd() { + let params = test_hyper_params("has_exited_running").unwrap(); + let mut hyper = HyperProcess::new(None, Some(¶ms)).unwrap(); + assert!( + !hyper.has_exited(), + "freshly-spawned hyperd should be running" + ); +} + +/// After SIGKILL, `has_exited` must observe the child as exited. +/// This is the path the daemon's monitor relies on to detect a dead hyperd. +/// Reaping the child as a side effect is also exercised here — without it, +/// the next `has_exited` call could return false on a zombie. +#[cfg(unix)] +#[test] +fn has_exited_returns_true_after_sigkill() { + use std::process::Command; + use std::time::Duration; + + let params = test_hyper_params("has_exited_killed").unwrap(); + let mut hyper = HyperProcess::new(None, Some(¶ms)).unwrap(); + let pid = hyper.pid().expect("hyperd should have a pid"); + + // Kill the process directly. The HyperProcess `Drop` would also kill it, + // but we need to test detection while we still own the handle. + let status = Command::new("kill") + .args(["-9", &pid.to_string()]) + .status() + .expect("kill -9 should succeed"); + assert!(status.success(), "kill -9 returned non-zero"); + + // Give the OS a moment to reap the SIGKILL'd process and update its state. + // Up to 2 seconds in 50ms increments — under load CI may take longer than + // a tight loop expects, but normal latency is sub-100ms. + let mut detected = false; + for _ in 0..40 { + if hyper.has_exited() { + detected = true; + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + assert!( + detected, + "has_exited should observe the killed child within 2s" + ); +} diff --git a/hyperdb-mcp/CHANGELOG.md b/hyperdb-mcp/CHANGELOG.md index d0cb8102..ed411b16 100644 --- a/hyperdb-mcp/CHANGELOG.md +++ b/hyperdb-mcp/CHANGELOG.md @@ -21,6 +21,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - Ephemeral databases now `DETACH DATABASE` before deletion on session end — required on Windows where the OS enforces file locks on open Hyper files. +- **Daemon-side `hyperd` restart on crash.** The daemon polls `hyperd` + every 5 seconds via `Child::try_wait()` and automatically restarts it + if the process has exited, atomically updating the discovery file + with the new endpoint. Clients reconnect transparently via the + existing `ConnectionLost` recovery path. New `REPORT_HYPERD_ERROR` + health-protocol command lets clients fast-path the signal when they + detect a dead hyperd before the daemon's polling tick. Restart + attempts are rate-limited to 3 per 60 seconds; exceeding the limit + triggers daemon shutdown so the user sees the failure clearly + rather than spinning silently. ## [0.1.1] - 2026-05-13 diff --git a/hyperdb-mcp/DEVELOPMENT.md b/hyperdb-mcp/DEVELOPMENT.md index d7ac0598..8204f97c 100644 --- a/hyperdb-mcp/DEVELOPMENT.md +++ b/hyperdb-mcp/DEVELOPMENT.md @@ -213,6 +213,34 @@ Cross-platform single-instance lock is the daemon's TCP health port — bind suc The daemon's main loop tracks idle time via `DaemonState::last_activity`. `HEARTBEAT` commands from active clients reset the timer; clients debounce these to once per 60 seconds in `HyperMcpServer::with_engine`. Idle timeout (default 30 min) triggers graceful shutdown: discovery file removed → `hyperd` dropped → health listener exits. +### hyperd liveness monitoring and restart + +A second monitor task in `run.rs::hyperd_monitor` polls the owned `HyperProcess` every 5 seconds via `HyperProcess::has_exited` (a `Child::try_wait()` call — zero SQL, correct on Unix and Windows, reaps zombies as a side effect). When the process is detected dead — or when a client sends `REPORT_HYPERD_ERROR` to the health port via `daemon::health::report_hyperd_error_to_daemon` — the monitor enters `try_restart_hyperd`: + +1. Prune the rolling restart-history vector to entries within `RESTART_WINDOW` (60s); reject if `RESTART_LIMIT` (3) attempts have already happened. +2. Drop the old `HyperProcess` (its `Drop` impl waits up to 5s for graceful shutdown — near-instant for an already-exited process). +3. Spawn a replacement via `HyperProcess::new` with the same parameters as initial startup. +4. Update the shared `Arc>` with the new endpoint (the health listener reads through the same Arc, so `STATUS` reports the current endpoint). +5. Atomically rewrite `daemon.json` via the existing temp-and-rename in `discovery::write_discovery_file`. + +After a successful restart, the monitor drains the restart-request flag once more — any `REPORT_HYPERD_ERROR` that landed *during* the restart was complaining about the now-replaced hyperd, not the freshly spawned one, and would otherwise trigger a spurious double-restart. + +When the rate limit is exceeded, the monitor returns; the main task observes that the `tokio::select!` branch completed, requests shutdown, and `hyperd_state` is dropped via Arc refcount as `run_daemon` returns. + +### Client-side recovery + +Existing engine recovery is unchanged: `HyperMcpServer::with_engine` drops the engine on `ConnectionLost` (server.rs around line 1085); the next tool call calls `Engine::new` → `try_daemon_mode` → re-reads `daemon.json` → connects to whatever endpoint is currently published. After a hyperd restart, the discovery file has the new endpoint, so reconnection happens transparently. + +Two new code paths fire `report_hyperd_error_to_daemon` (best-effort, 200ms timeouts so the user-facing tool handler isn't stalled): + +- After detecting `ConnectionLost` in `with_engine`. +- When `Connection::connect` to the daemon's advertised endpoint fails in `Engine::try_daemon_mode`. + +### Known limitations + +- **Hung-but-alive `hyperd`** (TCP listening, but unresponsive to queries) is NOT detected. The monitor's `try_wait()` returns `None` for a hung process; client tool calls hang on the read side without producing a `ConnectionLost` error. Operator recovery is `hyperdb-mcp daemon stop` followed by reconnect. +- **Watchers (background tasks holding `AsyncConnection` handles in `WatcherRegistry`)** do not currently auto-reconnect after a hyperd restart. They will go quiet until the user re-issues `watch_directory`. + See `src/daemon/{mod,discovery,health,run,spawn}.rs` for the full implementation. --- diff --git a/hyperdb-mcp/README.md b/hyperdb-mcp/README.md index 3a8b2ac2..6552289c 100644 --- a/hyperdb-mcp/README.md +++ b/hyperdb-mcp/README.md @@ -195,6 +195,16 @@ hyperdb-mcp daemon # Run as a daemon explicitly (rarely needed) State files live at `~/.hyperdb/` by default (override with `HYPERDB_STATE_DIR`). +### Recovery from hyperd crashes + +The daemon polls `hyperd` every 5 seconds. If the process has exited (crashed, OOM, killed), the daemon spawns a replacement, atomically updates `~/.hyperdb/daemon.json` with the new endpoint, and continues serving clients. Clients see one failed tool call (the request that was in flight when hyperd died); the next tool call transparently reconnects to the new hyperd via the same recovery path used for normal connection drops. + +If a client itself notices hyperd is unreachable before the next polling tick, it sends a fast-path `REPORT_HYPERD_ERROR` signal to the daemon so the restart kicks off without waiting for the timer. + +If hyperd repeatedly fails to start (3 attempts within 60 seconds — e.g., misconfigured `HYPERD_PATH`, port exhaustion, broken binary), the daemon shuts itself down and removes the discovery file. The next MCP client to start up will then spawn a fresh daemon, surfacing any persistent failure clearly to the user rather than spinning silently. + +**Known limitation:** if hyperd hangs (alive at the OS level but unresponsive to queries), the daemon's polling can't detect it and your tool call may stall indefinitely. The recovery path is `hyperdb-mcp daemon stop` followed by reconnecting from your MCP client. + ### Other behavioral flags | Flag | Behavior | diff --git a/hyperdb-mcp/src/daemon/health.rs b/hyperdb-mcp/src/daemon/health.rs index d9a4c695..15e11484 100644 --- a/hyperdb-mcp/src/daemon/health.rs +++ b/hyperdb-mcp/src/daemon/health.rs @@ -11,13 +11,16 @@ //! - `PING\n` → `PONG\n` (liveness check) //! - `HEARTBEAT\n` → `OK\n` (resets idle timer) //! - `STOP\n` → `STOPPING\n` (triggers graceful shutdown) -//! - `STATUS\n` → JSON line with daemon info +//! - `STATUS\n` → JSON line with daemon info (reports the *current* hyperd +//! endpoint, which can change after a restart). +//! - `REPORT_HYPERD_ERROR\n` → `OK\n` (sets the restart-requested flag — +//! the monitor task picks it up on its next tick). use std::io::{BufRead, BufReader, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; -use std::time::Instant; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use tracing::{debug, warn}; @@ -34,9 +37,12 @@ pub struct HealthListener { #[derive(Debug)] pub struct DaemonState { /// Last time any client sent a heartbeat or query. - pub last_activity: std::sync::Mutex, + pub last_activity: Mutex, /// Signal to shut down the daemon. pub shutdown: AtomicBool, + /// Set by clients reporting that hyperd looks dead from over there; + /// consumed by the daemon's restart monitor. + pub restart_requested: AtomicBool, } impl Default for DaemonState { @@ -48,8 +54,9 @@ impl Default for DaemonState { impl DaemonState { pub fn new() -> Self { Self { - last_activity: std::sync::Mutex::new(Instant::now()), + last_activity: Mutex::new(Instant::now()), shutdown: AtomicBool::new(false), + restart_requested: AtomicBool::new(false), } } @@ -65,7 +72,7 @@ impl DaemonState { /// /// # Panics /// Panics if the internal mutex is poisoned. - pub fn idle_duration(&self) -> std::time::Duration { + pub fn idle_duration(&self) -> Duration { self.last_activity.lock().expect("mutex poisoned").elapsed() } @@ -76,6 +83,17 @@ impl DaemonState { pub fn should_shutdown(&self) -> bool { self.shutdown.load(Ordering::Acquire) } + + /// Signal that hyperd appears to have died and a restart is needed. + pub fn request_restart(&self) { + self.restart_requested.store(true, Ordering::Release); + } + + /// Atomically read-and-clear the restart-request flag. + /// Returns true if a restart was requested since the last call. + pub fn consume_restart_request(&self) -> bool { + self.restart_requested.swap(false, Ordering::AcqRel) + } } impl HealthListener { @@ -94,11 +112,15 @@ impl HealthListener { /// Run the health listener loop. Spawns per-connection threads until shutdown. /// Consumes `self` because this is intended to be called from a dedicated thread. + /// + /// `info` is shared (`Arc>`) so the listener reports the + /// *current* hyperd endpoint after a restart — the monitor task updates the + /// same Arc once a new hyperd is running. #[expect( clippy::needless_pass_by_value, - reason = "Arc and DaemonInfo are cloned into per-connection threads" + reason = "Arcs are cloned into per-connection threads" )] - pub fn run(self, state: Arc, info: DaemonInfo) { + pub fn run(self, state: Arc, info: Arc>) { loop { if state.should_shutdown() { break; @@ -107,17 +129,17 @@ impl HealthListener { match self.listener.accept() { Ok((stream, _addr)) => { let state = Arc::clone(&state); - let info = info.clone(); + let info = Arc::clone(&info); std::thread::spawn(move || { handle_client(stream, &state, &info); }); } Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { - std::thread::sleep(std::time::Duration::from_millis(100)); + std::thread::sleep(Duration::from_millis(100)); } Err(e) => { warn!(error = %e, "health listener accept error"); - std::thread::sleep(std::time::Duration::from_millis(500)); + std::thread::sleep(Duration::from_millis(500)); } } } @@ -129,8 +151,8 @@ impl HealthListener { clippy::needless_pass_by_value, reason = "TcpStream must be owned for BufReader" )] -fn handle_client(stream: TcpStream, state: &DaemonState, info: &DaemonInfo) { - let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(5))); +fn handle_client(stream: TcpStream, state: &DaemonState, info: &Mutex) { + let _ = stream.set_read_timeout(Some(Duration::from_secs(5))); let mut reader = BufReader::new(&stream); let mut writer = &stream; let mut line = String::new(); @@ -152,9 +174,15 @@ fn handle_client(stream: TcpStream, state: &DaemonState, info: &DaemonInfo) { "STOPPING\n".to_string() } "STATUS" => { - let json = serde_json::to_string(info).unwrap_or_default(); + // Brief lock — only to clone the current snapshot. + let snapshot = info.lock().expect("DaemonInfo mutex poisoned").clone(); + let json = serde_json::to_string(&snapshot).unwrap_or_default(); format!("{json}\n") } + "REPORT_HYPERD_ERROR" => { + state.request_restart(); + "OK\n".to_string() + } _ => "ERR unknown command\n".to_string(), }; if writer.write_all(response.as_bytes()).is_err() { @@ -168,12 +196,52 @@ fn handle_client(stream: TcpStream, state: &DaemonState, info: &DaemonInfo) { /// Send a command to the daemon's health port and return the response. /// +/// Uses generous timeouts (2s connect, 5s read) suitable for `STOP`/`STATUS` +/// where the caller is willing to wait. Use [`send_command_with_timeout`] for +/// best-effort fire-and-forget calls (e.g. heartbeat, error reporting). +/// /// # Errors /// Returns an error if the connection fails or the response cannot be read. pub fn send_command(port: u16, command: &str) -> std::io::Result { + send_command_with_timeout( + port, + command, + Duration::from_secs(2), + Duration::from_secs(5), + ) +} + +/// Best-effort fire-and-forget: tell the running daemon that hyperd appears to +/// be dead from this client's perspective. Uses short timeouts (200ms each) so +/// the calling tool handler isn't stalled if the daemon itself is slow. +/// Errors are logged at debug level and otherwise ignored. +pub fn report_hyperd_error_to_daemon() { + let port = super::discovery::resolve_port(); + let timeout = Duration::from_millis(200); + match send_command_with_timeout(port, "REPORT_HYPERD_ERROR", timeout, timeout) { + Ok(response) => { + debug!(response = %response.trim(), "reported hyperd error to daemon"); + } + Err(e) => { + debug!(error = %e, "could not report hyperd error to daemon (best-effort)"); + } + } +} + +/// Send a command with caller-specified connect/read timeouts. +/// +/// # Errors +/// Returns an error if the connection fails or the response cannot be read +/// within the supplied timeouts. +pub fn send_command_with_timeout( + port: u16, + command: &str, + connect_timeout: Duration, + read_timeout: Duration, +) -> std::io::Result { let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port)); - let mut stream = TcpStream::connect_timeout(&addr, std::time::Duration::from_secs(2))?; - stream.set_read_timeout(Some(std::time::Duration::from_secs(5)))?; + let mut stream = TcpStream::connect_timeout(&addr, connect_timeout)?; + stream.set_read_timeout(Some(read_timeout))?; let msg = format!("{command}\n"); stream.write_all(msg.as_bytes())?; diff --git a/hyperdb-mcp/src/daemon/run.rs b/hyperdb-mcp/src/daemon/run.rs index 86b95a32..5224b06c 100644 --- a/hyperdb-mcp/src/daemon/run.rs +++ b/hyperdb-mcp/src/daemon/run.rs @@ -1,13 +1,14 @@ // Copyright (c) 2026, Salesforce, Inc. All rights reserved. // SPDX-License-Identifier: Apache-2.0 OR MIT -//! Daemon main loop: spawns `hyperd`, runs health listener, monitors idle timeout. +//! Daemon main loop: spawns `hyperd`, runs health listener, monitors idle timeout +//! and hyperd liveness, restarts hyperd if it dies. -use std::sync::Arc; -use std::time::Duration; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use tokio::signal; -use tracing::info; +use tracing::{error, info, warn}; use hyperdb_api::{HyperProcess, Parameters, TransportMode}; @@ -39,6 +40,29 @@ impl DaemonConfig { } } +/// Restart-attempt rate limit: at most 3 attempts within this window. +/// The 4th attempt within the window is rejected and triggers daemon shutdown. +pub const RESTART_WINDOW: Duration = Duration::from_secs(60); +pub const RESTART_LIMIT: usize = 3; + +/// Polling interval for the hyperd-liveness monitor. +const HYPERD_POLL_INTERVAL: Duration = Duration::from_secs(5); + +/// State the monitor task mutates and the main task drops on shutdown. +/// Holds the live `HyperProcess` and the rolling restart-attempt history. +struct HyperState { + hyper: Option, + restart_history: Vec, +} + +#[derive(Debug)] +enum RestartError { + /// More than `RESTART_LIMIT` restart attempts within `RESTART_WINDOW`. + TooManyRestarts, + /// `HyperProcess::new` failed or the new process produced no endpoint. + SpawnFailed(String), +} + /// Run the daemon. This function blocks until shutdown is triggered. /// /// # Errors @@ -60,24 +84,15 @@ pub async fn run_daemon(config: DaemonConfig) -> Result<(), Box Result<(), Box= idle_timeout { - info!( - idle_secs = idle_timeout.as_secs(), - "idle timeout reached, shutting down" - ); - shutdown_state.request_shutdown(); - break; - } - if shutdown_state.should_shutdown() { - break; - } - } - } => {} + () = idle_monitor(Arc::clone(&state), config.idle_timeout) => {} + () = hyperd_monitor(Arc::clone(&state), Arc::clone(&hyper_state), Arc::clone(&info_arc)) => {} () = shutdown_signal() => { info!("received shutdown signal"); - state.request_shutdown(); } } + state.request_shutdown(); - // Step 6: Graceful shutdown + // Step 7: Graceful shutdown. + // `tokio::select!` already cancelled the monitor and idle-monitor futures + // when one branch completed, releasing their `hyper_state` Arc clones. + // When this function returns, the last Arc drops, which drops the inner + // `HyperState`, which drops the `HyperProcess`, which closes the callback + // connection and lets hyperd exit cleanly. We don't lock-and-clear here + // because that would gain nothing — the same drop happens via Arc refcount. info!("shutting down daemon"); discovery::remove_discovery_file(); - drop(hyper); // closes callback connection → hyperd exits let _ = health_handle.join(); + drop(hyper_state); // explicit ordering: drop after health-listener join Ok(()) } +/// Outcome of a single rate-limit check on the restart-history vector. +#[derive(Debug, PartialEq, Eq)] +pub enum RestartAttempt { + /// The attempt is within the allowed budget; recorded. + Recorded, + /// `RESTART_LIMIT` attempts already happened in the current window. + LimitExceeded, +} + +/// Prune restart-history entries older than `RESTART_WINDOW`, then either +/// record `now` as a new attempt or report that the limit is already +/// exceeded. +/// +/// Pulled out as a standalone function so the rate-limit policy can be +/// tested directly without spinning up a real daemon. +pub fn try_record_restart_attempt(history: &mut Vec, now: Instant) -> RestartAttempt { + history.retain(|t| now.duration_since(*t) < RESTART_WINDOW); + if history.len() >= RESTART_LIMIT { + return RestartAttempt::LimitExceeded; + } + history.push(now); + RestartAttempt::Recorded +} + +/// Build the Parameters used for every hyperd spawn (initial start and restarts). +fn build_params() -> std::io::Result { + let log_dir = discovery::state_dir()?.join("logs"); + std::fs::create_dir_all(&log_dir)?; + + let mut params = Parameters::new(); + 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); + Ok(params) +} + +/// Watches for idle timeout. Triggers shutdown when no activity for +/// `idle_timeout`. Wakes every 10 seconds. +async fn idle_monitor(state: Arc, idle_timeout: Duration) { + loop { + tokio::time::sleep(Duration::from_secs(10)).await; + if state.should_shutdown() { + return; + } + if state.idle_duration() >= idle_timeout { + info!( + idle_secs = idle_timeout.as_secs(), + "idle timeout reached, shutting down" + ); + return; + } + } +} + +/// Watches hyperd's liveness. If hyperd has exited (or a client reported it as +/// dead), restarts it and rewrites the discovery file. If restarts exceed the +/// rate limit, returns and lets the main task initiate shutdown. +async fn hyperd_monitor( + state: Arc, + hyper_state: Arc>, + info_arc: Arc>, +) { + loop { + tokio::time::sleep(HYPERD_POLL_INTERVAL).await; + if state.should_shutdown() { + return; + } + + // Check liveness and consume the restart-request flag *atomically* with + // the decision: we only swap-to-false when we're actually committing to + // act, so a flag set after this point survives to the next tick. + let needs_restart = { + let mut guard = hyper_state.lock().expect("HyperState mutex poisoned"); + let process_dead = guard.hyper.as_mut().map_or(true, HyperProcess::has_exited); + // Only consume the flag when we're going to restart anyway, OR + // when the process is alive and we want to honor a client report. + if process_dead { + // Drain the flag so a stale post-death report doesn't cause a + // spurious double-restart on the next tick. + let _ = state.consume_restart_request(); + true + } else { + state.consume_restart_request() + } + }; + + if !needs_restart { + continue; + } + + match try_restart_hyperd(&hyper_state, &info_arc) { + Ok(new_endpoint) => { + info!(endpoint = %new_endpoint, "hyperd restarted"); + // Drain any reports that landed *during* the restart — those + // clients were complaining about the now-replaced hyperd, not + // the freshly spawned one. Without this, the next tick would + // see an alive process + a stale flag and trigger a spurious + // double-restart. + let _ = state.consume_restart_request(); + } + Err(RestartError::TooManyRestarts) => { + error!( + limit = RESTART_LIMIT, + window_secs = RESTART_WINDOW.as_secs(), + "hyperd restart limit exceeded — daemon shutting down" + ); + return; + } + Err(RestartError::SpawnFailed(e)) => { + warn!(error = %e, "hyperd spawn failed during restart; will retry on next tick"); + } + } + } +} + +/// Attempt one restart of hyperd. Drops the old process, spawns a new one, +/// updates `DaemonInfo`, and rewrites the discovery file. +/// +/// Every call (success or spawn-failure) consumes one slot from the rate-limit +/// window — a broken hyperd binary should not spin forever. +fn try_restart_hyperd( + hyper_state: &Mutex, + info_arc: &Mutex, +) -> Result { + let mut guard = hyper_state.lock().expect("HyperState mutex poisoned"); + + // Rate-limit check: prune-check-push. + if try_record_restart_attempt(&mut guard.restart_history, Instant::now()) + == RestartAttempt::LimitExceeded + { + return Err(RestartError::TooManyRestarts); + } + + // 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. + guard.hyper = None; + + // Spawn the replacement. + let params = build_params().map_err(|e| RestartError::SpawnFailed(e.to_string()))?; + let new_hyper = HyperProcess::new(None, Some(¶ms)) + .map_err(|e| RestartError::SpawnFailed(e.to_string()))?; + let new_endpoint = new_hyper + .endpoint() + .ok_or_else(|| RestartError::SpawnFailed("hyperd did not report endpoint".into()))? + .to_string(); + + // Publish the new endpoint to STATUS readers and to the discovery file. + // We snapshot the updated DaemonInfo while holding info_arc's lock, then + // write the file outside the lock to keep the critical section small. + let snapshot = { + let mut info_guard = info_arc.lock().expect("DaemonInfo mutex poisoned"); + info_guard.hyperd_endpoint.clone_from(&new_endpoint); + info_guard.clone() + }; + discovery::write_discovery_file(&snapshot) + .map_err(|e| RestartError::SpawnFailed(format!("discovery write: {e}")))?; + + guard.hyper = Some(new_hyper); + Ok(new_endpoint) +} + async fn shutdown_signal() { let ctrl_c = signal::ctrl_c(); diff --git a/hyperdb-mcp/src/engine.rs b/hyperdb-mcp/src/engine.rs index b225909b..29eef58f 100644 --- a/hyperdb-mcp/src/engine.rs +++ b/hyperdb-mcp/src/engine.rs @@ -193,6 +193,10 @@ impl Engine { CreateMode::CreateIfNotExists, ) .map_err(|e| { + // The daemon's discovery file points at this endpoint but we can't + // reach it — hyperd is likely dead. Tell the daemon so it can + // restart it on its next monitor tick. + daemon::health::report_hyperd_error_to_daemon(); McpError::new( ErrorCode::InternalError, format!("Failed to connect to daemon hyperd at {endpoint}: {e}"), diff --git a/hyperdb-mcp/src/server.rs b/hyperdb-mcp/src/server.rs index 0159eae7..3e24d0bf 100644 --- a/hyperdb-mcp/src/server.rs +++ b/hyperdb-mcp/src/server.rs @@ -1098,6 +1098,12 @@ impl HyperMcpServer { if let Ok(mut ready) = self.catalog_ready.lock() { *ready = false; } + // Tell the daemon hyperd looks dead from over here. The daemon + // will pick up the flag on its next monitor tick and restart. + // Skipped in --no-daemon mode because there's no daemon to tell. + if !self.no_daemon { + crate::daemon::health::report_hyperd_error_to_daemon(); + } } } result diff --git a/hyperdb-mcp/tests/daemon_tests.rs b/hyperdb-mcp/tests/daemon_tests.rs index 9af31d82..9bef11e9 100644 --- a/hyperdb-mcp/tests/daemon_tests.rs +++ b/hyperdb-mcp/tests/daemon_tests.rs @@ -49,6 +49,109 @@ fn daemon_state_default_is_equivalent_to_new() { assert!(default_state.idle_duration() < Duration::from_millis(100)); } +#[test] +fn daemon_state_default_initializes_restart_flag_false() { + // Guard against future regressions where Default and new() diverge — + // both must initialize restart_requested to false. + let default_state = DaemonState::default(); + let new_state = DaemonState::new(); + assert!(!default_state.consume_restart_request()); + assert!(!new_state.consume_restart_request()); +} + +#[test] +fn daemon_state_restart_request_consume_round_trip() { + let state = DaemonState::new(); + assert!(!state.consume_restart_request(), "initially clear"); + + state.request_restart(); + assert!(state.consume_restart_request(), "consume returns true once"); + assert!( + !state.consume_restart_request(), + "second consume returns false" + ); + + // Multiple requests coalesce into one consumption. + state.request_restart(); + state.request_restart(); + state.request_restart(); + assert!( + state.consume_restart_request(), + "three requests → one consume" + ); + assert!(!state.consume_restart_request()); +} + +#[test] +fn restart_history_records_attempts_under_limit() { + use hyperdb_mcp::daemon::run::{try_record_restart_attempt, RestartAttempt}; + let mut history: Vec = Vec::new(); + let t0 = Instant::now(); + + assert_eq!( + try_record_restart_attempt(&mut history, t0), + RestartAttempt::Recorded + ); + assert_eq!( + try_record_restart_attempt(&mut history, t0 + Duration::from_secs(10)), + RestartAttempt::Recorded + ); + assert_eq!( + try_record_restart_attempt(&mut history, t0 + Duration::from_secs(20)), + RestartAttempt::Recorded + ); + assert_eq!( + history.len(), + 3, + "first three attempts within window are recorded" + ); +} + +#[test] +fn restart_history_rejects_fourth_attempt_in_window() { + use hyperdb_mcp::daemon::run::{try_record_restart_attempt, RestartAttempt}; + let mut history: Vec = Vec::new(); + let t0 = Instant::now(); + + // Fill the window with 3 attempts. + try_record_restart_attempt(&mut history, t0); + try_record_restart_attempt(&mut history, t0 + Duration::from_secs(10)); + try_record_restart_attempt(&mut history, t0 + Duration::from_secs(20)); + + // 4th attempt within the 60s window must be rejected. + assert_eq!( + try_record_restart_attempt(&mut history, t0 + Duration::from_secs(30)), + RestartAttempt::LimitExceeded, + "4th attempt within window must be rejected" + ); + assert_eq!(history.len(), 3, "rejection must not push to history"); +} + +#[test] +fn restart_history_prunes_entries_older_than_window() { + use hyperdb_mcp::daemon::run::{try_record_restart_attempt, RestartAttempt}; + let mut history: Vec = Vec::new(); + let t0 = Instant::now(); + + // Three attempts at the start of the timeline. + try_record_restart_attempt(&mut history, t0); + try_record_restart_attempt(&mut history, t0 + Duration::from_secs(5)); + try_record_restart_attempt(&mut history, t0 + Duration::from_secs(10)); + + // Now jump 70 seconds — all three are stale and should be pruned. + let later = t0 + Duration::from_secs(70); + assert_eq!( + try_record_restart_attempt(&mut history, later), + RestartAttempt::Recorded, + "after window expires, restarts are allowed again" + ); + assert_eq!( + history.len(), + 1, + "stale entries pruned, only 'later' remains" + ); +} + // ─── Unit tests: Health protocol (no env vars, safe to run in parallel) ─────── #[test] @@ -121,6 +224,24 @@ fn health_protocol_unknown_command_returns_error() { assert!(response.contains("ERR")); } +#[test] +fn health_protocol_report_hyperd_error_sets_flag() { + let (port, _handle, state) = start_health_listener(); + + assert!(!state.consume_restart_request(), "flag starts clear"); + + let response = health::send_command(port, "REPORT_HYPERD_ERROR").unwrap(); + assert_eq!(response.trim(), "OK"); + + // The handler ran on a different thread; give it a moment to land the + // store. AcqRel ordering means the store is visible here as soon as the + // handler returns, but the response write is what unblocks send_command. + assert!( + state.consume_restart_request(), + "REPORT_HYPERD_ERROR must set the restart-requested flag" + ); +} + #[test] fn health_protocol_multi_command_session() { let (port, _handle, _state) = start_health_listener(); @@ -457,6 +578,110 @@ fn daemon_mode_persistent_engine_data_is_queryable() { assert_eq!(resp.trim(), "PONG"); } +#[cfg(unix)] +#[test] +fn hyperd_monitor_detects_killed_hyperd_and_restarts() { + let _lock = ENV_LOCK.lock().unwrap(); + let daemon = TestDaemon::start(); + + let pid_before = find_hyperd_pid_for_endpoint(&daemon.info.hyperd_endpoint) + .expect("should find hyperd pid for endpoint before kill"); + + // SIGKILL the hyperd process. The daemon's monitor should detect this on + // the next 5s tick and restart hyperd. + kill_pid(pid_before); + + // Wait up to 12 seconds for the monitor to fire and restart hyperd. + // (5s monitor tick + spawn time + slack.) + let new_endpoint = wait_for_endpoint_change_or_recovery(daemon.info.health_port, 12) + .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), + ); + assert!(probe.is_ok(), "new hyperd endpoint should be reachable"); +} + +#[cfg(unix)] +#[test] +fn client_report_triggers_restart_after_kill() { + let _lock = ENV_LOCK.lock().unwrap(); + let daemon = TestDaemon::start(); + + let pid_before = find_hyperd_pid_for_endpoint(&daemon.info.hyperd_endpoint) + .expect("should find hyperd pid before kill"); + kill_pid(pid_before); + + // Immediately tell the daemon hyperd is dead — don't wait for the monitor. + // Even so, the monitor only reacts on its 5s tick, so the worst-case + // recovery time is unchanged. This test just verifies the report path + // triggers the same restart as detection-via-polling. + let response = health::send_command(daemon.info.health_port, "REPORT_HYPERD_ERROR").unwrap(); + assert_eq!(response.trim(), "OK"); + + let new_endpoint = wait_for_endpoint_change_or_recovery(daemon.info.health_port, 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!(probe.is_ok(), "new hyperd endpoint should be reachable"); +} + +#[cfg(unix)] +#[test] +fn engine_recovers_after_hyperd_killed() { + // End-to-end test: the user-visible behavior of this whole feature. + // 1. Start daemon + create an Engine (= an MCP client connection). + // 2. Run a query — it succeeds. + // 3. SIGKILL hyperd. + // 4. Wait for the daemon to restart it. + // 5. Run another query through the same recovery path the server uses + // (drop engine on ConnectionLost, then create a fresh engine). + let _lock = ENV_LOCK.lock().unwrap(); + let daemon = TestDaemon::start(); + + let tmp = TempDir::new().unwrap(); + let path = tmp.path().join("recover.hyper"); + let path_str = path.to_str().unwrap().to_string(); + + // Engine #1: pre-kill + { + let engine = hyperdb_mcp::engine::Engine::new(Some(path_str.clone())).unwrap(); + engine + .execute_command("CREATE TABLE keepers (n INT)") + .unwrap(); + engine + .execute_command("INSERT INTO keepers VALUES (1), (2), (3)") + .unwrap(); + } + + // Find and kill hyperd + let pid = + find_hyperd_pid_for_endpoint(&daemon.info.hyperd_endpoint).expect("should find hyperd pid"); + kill_pid(pid); + + // Wait for daemon-side restart (discovery file gets a new endpoint). + wait_for_endpoint_change_or_recovery(daemon.info.health_port, 12) + .expect("daemon should restart hyperd within 12s"); + + // Engine #2: post-restart. This mirrors what `with_engine` does after a + // ConnectionLost — drop the old engine (already done above) and create a + // fresh one. The fresh engine re-discovers the daemon and connects to the + // new endpoint. + let engine = hyperdb_mcp::engine::Engine::new(Some(path_str)).unwrap(); + // The persistent .hyper file is intact on disk. Re-attaching via a new + // engine should let us see the data we wrote pre-kill. + let rows = engine + .execute_query_to_json("SELECT n FROM keepers ORDER BY n") + .unwrap(); + assert_eq!(rows.len(), 3, "data persisted across hyperd restart"); +} + #[test] fn daemon_mode_ephemeral_database_cleaned_up_on_drop() { let _lock = ENV_LOCK.lock().unwrap(); @@ -489,13 +714,13 @@ fn start_health_listener() -> (u16, std::thread::JoinHandle<()>, Arc u16 { listener.local_addr().unwrap().port() } +/// 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. +#[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()?; + if !output.status.success() { + return None; + } + output + .stdout + .split(|b| *b == b'\n') + .filter_map(|line| std::str::from_utf8(line).ok()) + .map(str::trim) + .find(|s| !s.is_empty()) + .and_then(|s| s.parse::().ok()) +} + +/// Kill the given PID with SIGKILL. Unix-only. +#[cfg(unix)] +fn kill_pid(pid: u32) { + let status = std::process::Command::new("kill") + .args(["-9", &pid.to_string()]) + .status() + .expect("kill -9 should run"); + assert!(status.success(), "kill -9 {pid} failed"); +} + +/// Poll the daemon's `STATUS` endpoint until the reported `hyperd_endpoint` is +/// reachable (i.e. a fresh hyperd has been spawned after a kill). Returns the +/// endpoint string, or `None` if the timeout expires. +#[cfg(unix)] +fn wait_for_endpoint_change_or_recovery(health_port: u16, timeout_secs: u64) -> Option { + let deadline = Instant::now() + Duration::from_secs(timeout_secs); + while Instant::now() < deadline { + if let Ok(response) = health::send_command(health_port, "STATUS") { + if let Ok(parsed) = serde_json::from_str::(response.trim()) { + if let Some(endpoint) = parsed["hyperd_endpoint"].as_str() { + if let Ok(addr) = endpoint.parse::() { + if std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(500)) + .is_ok() + { + return Some(endpoint.to_string()); + } + } + } + } + } + std::thread::sleep(Duration::from_millis(250)); + } + None +} + /// RAII guard that sets/removes an environment variable and restores it on drop. struct EnvGuard { key: String,