use std::io;
use std::process::Command;
use std::time::{Duration, Instant};
use tracing::{debug, info, warn};
use super::discovery::{self, DaemonInfo, PortScan, ScanOutcome};
const SPAWN_TIMEOUT: Duration = Duration::from_secs(10);
const POLL_INTERVAL: Duration = Duration::from_millis(100);
pub fn ensure_daemon(scan: PortScan) -> io::Result<DaemonInfo> {
if let Some(info) = discovery::discover() {
return maybe_take_over(info, scan);
}
match discovery::scan_for_daemon(scan) {
ScanOutcome::Found(info) => maybe_take_over(*info, scan),
ScanOutcome::FreePort(port) => {
info!(port, "no running daemon detected, spawning on free port");
spawn_detached(port)?;
let info = wait_for_daemon()?;
if info.health_port > scan.base {
let lower_scan = PortScan {
base: scan.base,
span: info.health_port.saturating_sub(scan.base),
};
if let ScanOutcome::Found(lower_info) = discovery::scan_for_daemon(lower_scan) {
debug!(
prefer_port = lower_info.health_port,
stop_port = info.health_port,
"found lower-port daemon from concurrent spawn; stopping off-base daemon"
);
let _ = super::health::send_command(info.health_port, "STOP");
return Ok(*lower_info);
}
}
Ok(info)
}
ScanOutcome::AllOccupied => Err(io::Error::new(
io::ErrorKind::AddrInUse,
format!(
"no free hyperdb daemon port in {}..{}",
scan.base,
scan.base.saturating_add(scan.span)
),
)),
}
}
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);
cmd.stdin(std::process::Stdio::null());
cmd.stdout(std::process::Stdio::null());
cmd.stderr(std::process::Stdio::null());
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
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(())
}
fn wait_for_daemon() -> io::Result<DaemonInfo> {
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);
}
}
pub fn client_should_take_over(client_ver: &str, daemon_ver: &str) -> bool {
let Ok(client) = semver::Version::parse(client_ver) else {
return false;
};
let Ok(daemon) = semver::Version::parse(daemon_ver) else {
return false;
};
client > daemon
}
fn maybe_take_over(info: DaemonInfo, _scan: PortScan) -> io::Result<DaemonInfo> {
let client_ver = crate::version::MCP_VERSION;
if !client_should_take_over(client_ver, &info.version) {
let parse_failed = semver::Version::parse(client_ver).is_err()
|| semver::Version::parse(&info.version).is_err();
debug!(
daemon_version = %info.version,
client_version = %client_ver,
port = info.health_port,
reason = if parse_failed { "version unparseable" } else { "client not newer" },
"reusing existing daemon"
);
return Ok(info);
}
info!(
daemon_version = %info.version,
client_version = %client_ver,
port = info.health_port,
"newer MCP client taking over older daemon"
);
let _ = super::health::send_command(info.health_port, "STOP");
let deadline = Instant::now() + SPAWN_TIMEOUT;
while Instant::now() < deadline {
if super::health::ping_identified(
info.health_port,
Duration::from_millis(200),
Duration::from_millis(200),
)
.is_none()
{
if let Some(fresh) = discovery::discover() {
if fresh.health_port == info.health_port {
return Ok(fresh);
}
}
spawn_detached(info.health_port)?;
return wait_for_daemon();
}
std::thread::sleep(POLL_INTERVAL);
}
warn!(
port = info.health_port,
timeout_secs = SPAWN_TIMEOUT.as_secs(),
"old daemon did not stop within timeout, reusing it"
);
Ok(info)
}