use rmcp::ServiceExt;
use velesdb_memory::mcp::McpServer;
pub(crate) async fn serve_stdio(
server: McpServer,
original_parent: u32,
) -> Result<(), Box<dyn std::error::Error>> {
spawn_orphan_watchdog(original_parent);
let running = server
.serve((tokio::io::stdin(), tokio::io::stdout()))
.await?;
running.waiting().await?;
Ok(())
}
#[cfg(feature = "http")]
pub(crate) async fn serve_selected_transport(
server: McpServer,
http_bind: Option<HttpServeRequest>,
original_parent: u32,
) -> Result<(), Box<dyn std::error::Error>> {
if let Some(request) = http_bind {
return serve_http(server, request).await;
}
serve_stdio(server, original_parent).await
}
#[cfg(not(feature = "http"))]
pub(crate) async fn serve_selected_transport(
server: McpServer,
_http_bind: Option<String>,
original_parent: u32,
) -> Result<(), Box<dyn std::error::Error>> {
serve_stdio(server, original_parent).await
}
pub(crate) fn apply_logging() {
if let Err(err) = velesdb_memory::logging::init_from_env() {
eprintln!("[velesdb-memory] {err}");
std::process::exit(1);
}
}
#[cfg(feature = "http")]
pub(crate) struct HttpServeRequest {
bind_addr: String,
insecure: bool,
}
#[cfg(feature = "http")]
pub(crate) fn requested_http_bind(args: &[String]) -> Option<HttpServeRequest> {
let http_flag = args.iter().any(|arg| arg == "--http");
let http_env = std::env::var("VELESDB_MEMORY_HTTP").as_deref() == Ok("1");
if !http_flag && !http_env {
return None;
}
let port_override = args
.iter()
.position(|arg| arg == "--http-port")
.and_then(|flag_index| args.get(flag_index + 1));
let default_bind = std::env::var("VELESDB_MEMORY_HTTP_BIND")
.unwrap_or_else(|_| velesdb_memory::http::DEFAULT_HTTP_BIND.to_owned());
let bind_addr = match port_override {
Some(port) => match default_bind.rsplit_once(':') {
Some((host, _existing_port)) => format!("{host}:{port}"),
None => format!("127.0.0.1:{port}"),
},
None => default_bind,
};
if !is_loopback_host(&bind_addr)
&& std::env::var("VELESDB_MEMORY_HTTP_ALLOW_REMOTE").as_deref() != Ok("1")
{
eprintln!(
"[velesdb-memory] refusing to bind the HTTP transport to '{bind_addr}': it is not a \
loopback address, and the streamable-HTTP transport has no authentication — anyone \
who can reach that socket gets full read/write access to the store. Set \
VELESDB_MEMORY_HTTP_ALLOW_REMOTE=1 to override (put an authenticating reverse proxy \
in front first)."
);
std::process::exit(1);
}
let insecure_flag = args.iter().any(|arg| arg == "--http-insecure");
let insecure_env = std::env::var("VELESDB_MEMORY_HTTP_INSECURE").as_deref() == Ok("1");
let insecure = insecure_flag || insecure_env;
Some(HttpServeRequest {
bind_addr,
insecure,
})
}
#[cfg(feature = "http")]
pub(crate) fn is_loopback_host(bind_addr: &str) -> bool {
let host = bind_addr
.rsplit_once(':')
.map_or(bind_addr, |(host, _port)| host)
.trim_start_matches('[')
.trim_end_matches(']');
host.parse::<std::net::IpAddr>()
.is_ok_and(|ip| ip.is_loopback())
}
#[cfg(not(feature = "http"))]
pub(crate) fn requested_http_bind(args: &[String]) -> Option<String> {
let http_flag = args.iter().any(|arg| arg == "--http");
let http_env = std::env::var("VELESDB_MEMORY_HTTP").as_deref() == Ok("1");
if http_flag || http_env {
eprintln!(
"[velesdb-memory] --http / VELESDB_MEMORY_HTTP=1 requires a binary built with \
`--features http` (e.g. `cargo install velesdb-memory --features http`) — \
this binary was built without it"
);
std::process::exit(1);
}
None
}
#[cfg(feature = "http")]
pub(crate) async fn serve_http(
server: McpServer,
request: HttpServeRequest,
) -> Result<(), Box<dyn std::error::Error>> {
let HttpServeRequest {
bind_addr,
insecure,
} = request;
let ct = tokio_util::sync::CancellationToken::new();
let app = velesdb_memory::http::router(server, ct.child_token());
let listener = tokio::net::TcpListener::bind(&bind_addr).await?;
spawn_shutdown_signals(ct.clone());
if insecure {
eprintln!(
"[velesdb-memory] WARNING: --http-insecure / VELESDB_MEMORY_HTTP_INSECURE=1 is set — \
serving PLAIN HTTP (no TLS) on http://{bind_addr}/mcp. Every request is readable by \
anyone who can reach that socket (loopback-only by default — see \
VELESDB_MEMORY_HTTP_ALLOW_REMOTE above). Use this only for local debugging, or when \
a trusted TLS-terminating proxy already sits in front."
);
eprintln!("[velesdb-memory] HTTP server listening on http://{bind_addr}/mcp");
axum::serve(listener, app)
.with_graceful_shutdown(async move { ct.cancelled_owned().await })
.await?;
return Ok(());
}
let tls_dir = velesdb_memory::tls::tls_dir_from_env();
let material = velesdb_memory::tls::ensure_tls_material(&tls_dir)?;
let acceptor = velesdb_memory::tls::tls_acceptor_from_material(&material)?;
eprintln!("[velesdb-memory] HTTPS server listening on https://{bind_addr}/mcp");
eprintln!(
"[velesdb-memory] Local CA: {} — a client only needs to trust this once (see \
./scripts/install-memory-daemon.sh, which does this automatically on macOS); every \
future leaf certificate this daemon issues is signed by the same CA and is trusted \
automatically after that.",
material.ca_cert_path.display()
);
velesdb_memory::http::serve_tls(app, listener, acceptor, ct).await;
Ok(())
}
#[cfg(unix)]
pub(crate) const ORPHAN_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);
#[cfg(unix)]
pub(crate) fn spawn_orphan_watchdog(original_parent: u32) {
use std::os::unix::process::parent_id;
tokio::spawn(async move {
loop {
tokio::time::sleep(ORPHAN_CHECK_INTERVAL).await;
let current_parent = parent_id();
if current_parent != original_parent {
eprintln!(
"[velesdb-memory] parent process (pid {original_parent}) is gone \
(now reparented under pid {current_parent}) — exiting to release \
the store lock rather than leak a zombie session (#1448)"
);
std::process::exit(0);
}
}
});
}
#[cfg(not(unix))]
pub(crate) fn spawn_orphan_watchdog(_original_parent: u32) {}
#[cfg(feature = "http")]
pub(crate) fn spawn_shutdown_signals(ct: tokio_util::sync::CancellationToken) {
let interrupt = ct.clone();
tokio::spawn(async move {
if tokio::signal::ctrl_c().await.is_ok() {
interrupt.cancel();
}
});
#[cfg(unix)]
tokio::spawn(async move {
if let Ok(mut term) =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
{
term.recv().await;
ct.cancel();
}
});
#[cfg(not(unix))]
drop(ct);
}
#[cfg(all(test, feature = "http"))]
#[path = "daemon_serve_tests.rs"]
mod tests;