#![cfg(any(unix, windows))]
use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};
use fs2::FileExt;
use serde::{Deserialize, Serialize};
use crate::store_layout::{cache_root, workspace_key};
const DAEMON_LOCK_FILE: &str = "daemon.lock";
const DAEMON_PID_FILE: &str = "daemon.pid";
const DAEMONS_SUBDIR: &str = "daemons";
pub const MAX_LIVE_DAEMONS: usize = 8;
pub const MAX_LIVE_DAEMONS_ENV: &str = "BASEMIND_MAX_DAEMONS";
pub fn max_live_daemons() -> usize {
match std::env::var(MAX_LIVE_DAEMONS_ENV) {
Ok(raw) => match raw.trim().parse::<usize>() {
Ok(n) if n > 0 => n,
_ => MAX_LIVE_DAEMONS,
},
Err(_) => MAX_LIVE_DAEMONS,
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum DaemonKind {
#[default]
Comms,
Agent,
Shells,
}
impl DaemonKind {
pub fn as_str(self) -> &'static str {
match self {
Self::Comms => "comms",
Self::Agent => "agent",
Self::Shells => "shells",
}
}
}
impl std::fmt::Display for DaemonKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaemonRecord {
pub pid: u32,
#[serde(default)]
pub kind: DaemonKind,
#[serde(alias = "comms_dir")]
pub dir: PathBuf,
pub version: String,
pub started_unix: i64,
}
fn daemons_dir() -> PathBuf {
cache_root().join(DAEMONS_SUBDIR)
}
fn machine_registry_path_in(machine_dir: &Path, kind: DaemonKind, dir: &Path) -> PathBuf {
machine_dir.join(format!("{}-{}.pid", kind.as_str(), workspace_key(dir)))
}
#[cfg(unix)]
pub fn pid_is_live(pid: u32) -> bool {
let rc = unsafe { libc::kill(pid as libc::pid_t, 0) };
if rc == 0 {
return true;
}
std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
}
#[cfg(windows)]
pub fn pid_is_live(pid: u32) -> bool {
const PROCESS_QUERY_LIMITED_INFORMATION: u32 = 0x1000;
unsafe {
let handle = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid);
if handle == 0 {
return false;
}
CloseHandle(handle);
true
}
}
#[cfg(windows)]
#[link(name = "kernel32")]
unsafe extern "system" {
fn OpenProcess(access: u32, inherit: i32, pid: u32) -> isize;
fn CloseHandle(handle: isize) -> i32;
}
fn read_record(path: &Path) -> Option<DaemonRecord> {
serde_json::from_slice(&std::fs::read(path).ok()?).ok()
}
pub fn live_daemons() -> Vec<DaemonRecord> {
live_daemons_in(&daemons_dir())
}
pub fn live_daemons_of(kind: DaemonKind) -> Vec<DaemonRecord> {
live_daemons_of_in(&daemons_dir(), kind)
}
fn live_daemons_of_in(machine_dir: &Path, kind: DaemonKind) -> Vec<DaemonRecord> {
let mut live = live_daemons_in(machine_dir);
live.retain(|record| record.kind == kind);
live
}
fn live_daemons_in(machine_dir: &Path) -> Vec<DaemonRecord> {
let Ok(entries) = std::fs::read_dir(machine_dir) else {
return Vec::new();
};
let mut live = Vec::new();
for entry in entries.flatten() {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("pid") {
continue;
}
match read_record(&path) {
Some(record) if pid_is_live(record.pid) => live.push(record),
_ => {
let _ = std::fs::remove_file(&path);
}
}
}
live
}
pub fn count_live_daemons_of(kind: DaemonKind) -> usize {
live_daemons_of(kind).len()
}
#[derive(Debug)]
pub struct DaemonLock {
_lock: File,
pid_path: PathBuf,
machine_path: PathBuf,
}
#[derive(Debug)]
pub enum DaemonLockOutcome {
Acquired(DaemonLock),
AlreadyHeld(Option<DaemonRecord>),
}
impl DaemonLock {
pub fn acquire(comms_dir: &Path, version: &str) -> std::io::Result<DaemonLockOutcome> {
Self::acquire_kind(DaemonKind::Comms, comms_dir, version)
}
pub fn acquire_kind(kind: DaemonKind, dir: &Path, version: &str) -> std::io::Result<DaemonLockOutcome> {
Self::acquire_at(kind, dir, version, &daemons_dir())
}
pub fn acquire_at(
kind: DaemonKind,
dir: &Path,
version: &str,
machine_dir: &Path,
) -> std::io::Result<DaemonLockOutcome> {
let lock_path = dir.join(DAEMON_LOCK_FILE);
let file = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(&lock_path)?;
if file.try_lock_exclusive().is_err() {
let pid_path = dir.join(DAEMON_PID_FILE);
return Ok(DaemonLockOutcome::AlreadyHeld(read_record(&pid_path)));
}
let record = DaemonRecord {
pid: std::process::id(),
kind,
dir: dir.to_path_buf(),
version: version.to_string(),
started_unix: now_unix(),
};
let pid_path = dir.join(DAEMON_PID_FILE);
let machine_path = machine_registry_path_in(machine_dir, kind, dir);
write_record(&pid_path, &record);
let _ = std::fs::create_dir_all(machine_dir);
write_record(&machine_path, &record);
Ok(DaemonLockOutcome::Acquired(DaemonLock {
_lock: file,
pid_path,
machine_path,
}))
}
}
impl Drop for DaemonLock {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.pid_path);
let _ = std::fs::remove_file(&self.machine_path);
}
}
fn write_record(path: &Path, record: &DaemonRecord) {
let Ok(bytes) = serde_json::to_vec(record) else {
return;
};
let tmp = path.with_extension(format!("pid.{}.tmp", std::process::id()));
if std::fs::write(&tmp, &bytes).is_ok() {
let _ = std::fs::rename(&tmp, path);
}
}
fn now_unix() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn self_pid_is_live_and_a_bogus_pid_is_not() {
assert!(pid_is_live(std::process::id()), "our own pid is live");
assert!(!pid_is_live(0x7FFF_FFFE), "an absent pid is not live");
}
#[test]
fn a_record_written_by_an_older_binary_still_parses_as_a_comms_daemon() {
let old_shape = br#"{"pid":4242,"comms_dir":"/tmp/old-comms","version":"0.22.0","started_unix":17}"#;
let record: DaemonRecord = serde_json::from_slice(old_shape).expect("an old-shape record must still parse");
assert_eq!(record.kind, DaemonKind::Comms, "an untagged record is a comms daemon");
assert_eq!(
record.dir,
PathBuf::from("/tmp/old-comms"),
"the directory survives the rename"
);
assert_eq!(record.pid, 4242);
}
#[test]
fn acquire_admits_one_owner_then_reports_already_held() {
let machine = tempfile::tempdir().expect("machine tempdir");
let comms = tempfile::tempdir().expect("comms tempdir");
let first =
DaemonLock::acquire_at(DaemonKind::Comms, comms.path(), "9.9.9", machine.path()).expect("first acquire");
assert!(
matches!(first, DaemonLockOutcome::Acquired(_)),
"the first daemon wins the lock"
);
match DaemonLock::acquire_at(DaemonKind::Comms, comms.path(), "9.9.9", machine.path()).expect("second acquire")
{
DaemonLockOutcome::AlreadyHeld(Some(record)) => {
assert_eq!(record.pid, std::process::id(), "the holder record names us");
}
other => panic!("a second acquire must report AlreadyHeld with the holder, got {other:?}"),
}
}
#[test]
fn count_prunes_dead_holders_and_counts_live_ones() {
let machine = tempfile::tempdir().expect("machine tempdir");
let comms = tempfile::tempdir().expect("comms tempdir");
let held = DaemonLock::acquire_at(DaemonKind::Comms, comms.path(), "9.9.9", machine.path()).expect("acquire");
assert!(matches!(held, DaemonLockOutcome::Acquired(_)));
assert_eq!(
live_daemons_in(machine.path()).len(),
1,
"our live daemon is counted once"
);
let dead = DaemonRecord {
pid: 0x7FFF_FFFE,
kind: DaemonKind::Comms,
dir: PathBuf::from("/nonexistent"),
version: "9.9.9".to_string(),
started_unix: now_unix(),
};
let dead_path = machine.path().join("dead.pid");
write_record(&dead_path, &dead);
assert_eq!(
live_daemons_in(machine.path()).len(),
1,
"the dead holder is pruned, only the live one counts"
);
assert!(!dead_path.exists(), "the dead pidfile was reaped");
}
#[test]
fn the_ceiling_counts_each_kind_separately() {
let machine = tempfile::tempdir().expect("machine tempdir");
let comms = tempfile::tempdir().expect("comms tempdir");
let agent = tempfile::tempdir().expect("agent tempdir");
let _comms_lock =
DaemonLock::acquire_at(DaemonKind::Comms, comms.path(), "9.9.9", machine.path()).expect("comms acquire");
let _agent_lock =
DaemonLock::acquire_at(DaemonKind::Agent, agent.path(), "9.9.9", machine.path()).expect("agent acquire");
assert_eq!(live_daemons_in(machine.path()).len(), 2, "both families are registered");
let comms_only = live_daemons_of_in(machine.path(), DaemonKind::Comms);
assert_eq!(comms_only.len(), 1, "an agent daemon does not count against comms");
assert_eq!(comms_only[0].dir, comms.path(), "the comms entry names the comms dir");
assert_eq!(
live_daemons_of_in(machine.path(), DaemonKind::Shells).len(),
0,
"no shells daemon is registered"
);
}
#[test]
fn drop_releases_the_lock_and_removes_the_registry_entry() {
let machine = tempfile::tempdir().expect("machine tempdir");
let comms = tempfile::tempdir().expect("comms tempdir");
{
let held =
DaemonLock::acquire_at(DaemonKind::Comms, comms.path(), "9.9.9", machine.path()).expect("acquire");
assert!(matches!(held, DaemonLockOutcome::Acquired(_)));
assert_eq!(live_daemons_in(machine.path()).len(), 1);
}
assert_eq!(
live_daemons_in(machine.path()).len(),
0,
"dropping the lock deregisters the daemon"
);
match DaemonLock::acquire_at(DaemonKind::Comms, comms.path(), "9.9.9", machine.path())
.expect("re-acquire after drop")
{
DaemonLockOutcome::Acquired(_) => {}
other => panic!("the lock is free after drop, got {other:?}"),
}
}
}