use std::fs::{File, OpenOptions};
use std::io;
use std::path::Path;
use fs4::fs_std::FileExt;
use serde::{Deserialize, Serialize};
pub use crate::events::registry::{daemon_paths, daemon_paths_in, DaemonPaths};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct DaemonInfo {
pub pid: u32,
pub protocol_version: u32,
pub daemon_version: String,
pub profile: Option<String>,
pub started_at: chrono::DateTime<chrono::Utc>,
pub socket: String,
#[serde(default)]
pub exe: Option<std::path::PathBuf>,
#[serde(default = "one")]
pub generation: u64,
}
fn one() -> u64 {
1
}
#[cfg(unix)]
fn chmod_600(path: &Path) {
use std::os::unix::fs::PermissionsExt;
let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600));
}
#[cfg(not(unix))]
fn chmod_600(_path: &Path) {}
pub fn write_daemon_json(paths: &DaemonPaths, info: &DaemonInfo) -> io::Result<()> {
let body = serde_json::to_vec_pretty(info).map_err(io::Error::other)?;
write_atomic(&paths.json, &body)?;
write_atomic(&paths.pid, format!("{}\n", info.pid).as_bytes())
}
fn temp_path(path: &Path) -> std::path::PathBuf {
let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
name.push(".tmp");
path.with_file_name(name)
}
fn create_private(path: &Path) -> io::Result<File> {
let mut options = OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600).custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);
}
options.open(path)
}
#[cfg(unix)]
fn set_private(file: &File) -> io::Result<()> {
use std::os::unix::fs::PermissionsExt;
file.set_permissions(std::fs::Permissions::from_mode(0o600))
}
#[cfg(not(unix))]
fn set_private(_file: &File) -> io::Result<()> {
Ok(())
}
fn write_atomic(path: &Path, body: &[u8]) -> io::Result<()> {
use std::io::Write;
let tmp = temp_path(path);
let mut file = match create_private(&tmp) {
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
std::fs::remove_file(&tmp)?;
create_private(&tmp)?
}
opened => opened?,
};
let written = file.write_all(body).and_then(|()| set_private(&file));
drop(file);
#[cfg(test)]
test_pause::at(&test_pause::BEFORE_RENAME);
let renamed = written.and_then(|()| std::fs::rename(&tmp, path));
if renamed.is_err() {
let _ = std::fs::remove_file(&tmp);
}
renamed
}
pub fn write_private_atomic(path: &Path, body: &[u8]) -> io::Result<()> {
write_atomic(path, body)
}
pub fn read_daemon_json(paths: &DaemonPaths) -> Option<DaemonInfo> {
let s = std::fs::read_to_string(&paths.json).ok()?;
serde_json::from_str(&s).ok()
}
pub fn heal_daemon_json(paths: &DaemonPaths, info: &DaemonInfo, _lock: &DaemonLock) -> io::Result<bool> {
match std::fs::symlink_metadata(&paths.json) {
Ok(m) if !m.file_type().is_file() => return Ok(false),
Ok(_) if read_daemon_json(paths).is_some_and(|on_disk| on_disk.pid == info.pid) => return Ok(false),
Ok(_) => {}
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
Err(e) => return Err(e),
}
#[cfg(test)]
test_pause::at(&test_pause::HEAL_BEFORE_WRITE);
write_daemon_json(paths, info)?;
Ok(true)
}
pub const JSON_HEAL_LOG_EVERY: std::time::Duration = std::time::Duration::from_secs(600);
pub struct Metadata {
paths: DaemonPaths,
inner: std::sync::Mutex<MetaInner>,
}
struct MetaInner {
info: DaemonInfo,
heals_logged_at: Option<std::time::Instant>,
heals_since_log: u64,
cleaned: bool,
}
#[derive(Debug, PartialEq, Eq)]
pub enum Healed {
No,
Yes { info: DaemonInfo, log: Option<u64> },
}
impl Metadata {
pub fn start(paths: DaemonPaths, info: DaemonInfo) -> io::Result<Self> {
let meta = Self {
paths,
inner: std::sync::Mutex::new(MetaInner {
info,
heals_logged_at: None,
heals_since_log: 0,
cleaned: false,
}),
};
{
let inner = meta.inner();
write_daemon_json(&meta.paths, &inner.info)?;
}
Ok(meta)
}
fn inner(&self) -> std::sync::MutexGuard<'_, MetaInner> {
self.inner.lock().unwrap_or_else(|e| e.into_inner())
}
pub fn info(&self) -> DaemonInfo {
self.inner().info.clone()
}
pub fn publish(&self, info: DaemonInfo) -> io::Result<()> {
let mut inner = self.inner();
inner.info = info;
if inner.cleaned {
return Ok(());
}
write_daemon_json(&self.paths, &inner.info)
}
pub fn heal(&self, lock: &DaemonLock) -> io::Result<Healed> {
let mut inner = self.inner();
if inner.cleaned || !heal_daemon_json(&self.paths, &inner.info, lock)? {
return Ok(Healed::No);
}
inner.heals_since_log += 1;
let log = if inner.heals_logged_at.is_none_or(|at| at.elapsed() >= JSON_HEAL_LOG_EVERY) {
let heals = inner.heals_since_log;
inner.heals_logged_at = Some(std::time::Instant::now());
inner.heals_since_log = 0;
Some(heals)
} else {
None
};
Ok(Healed::Yes { info: inner.info.clone(), log })
}
pub fn cleanup(&self, lock: &DaemonLock) -> bool {
let mut inner = self.inner();
inner.cleaned = true;
cleanup_metadata(&self.paths, lock)
}
}
#[cfg(test)]
pub(crate) mod test_pause {
use std::cell::RefCell;
use std::thread::LocalKey;
pub(crate) type Hook = RefCell<Option<Box<dyn Fn()>>>;
thread_local! {
pub(crate) static HEAL_BEFORE_WRITE: Hook = const { RefCell::new(None) };
pub(crate) static BEFORE_RENAME: Hook = const { RefCell::new(None) };
}
pub(crate) fn set(slot: &'static LocalKey<Hook>, hook: impl Fn() + 'static) {
slot.with(|h| *h.borrow_mut() = Some(Box::new(hook)));
}
pub(crate) fn at(slot: &'static LocalKey<Hook>) {
slot.with(|h| {
if let Some(hook) = h.borrow().as_ref() {
hook();
}
});
}
}
pub struct DaemonLock {
file: File,
}
impl DaemonLock {
pub fn try_acquire(paths: &DaemonPaths) -> io::Result<Option<Self>> {
let file = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&paths.lock)?;
chmod_600(&paths.lock);
if FileExt::try_lock_exclusive(&file)? {
Ok(Some(Self { file }))
} else {
Ok(None)
}
}
}
impl DaemonLock {
#[cfg(unix)]
pub fn adopt(fd: i32) -> io::Result<Self> {
use std::os::unix::io::FromRawFd;
if fd < 0 {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "negative lock fd"));
}
let file = unsafe { File::from_raw_fd(fd) };
unsafe {
let flags = libc::fcntl(fd, libc::F_GETFD);
if flags >= 0 {
libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC);
}
}
Ok(Self { file })
}
#[cfg(unix)]
pub fn raw_fd(&self) -> i32 {
use std::os::unix::io::AsRawFd;
self.file.as_raw_fd()
}
}
impl Drop for DaemonLock {
fn drop(&mut self) {
let _ = FileExt::unlock(&self.file);
}
}
impl DaemonPaths {
pub fn spawn_lock(&self) -> std::path::PathBuf {
self.lock.with_extension("spawn.lock")
}
}
#[derive(Debug)]
pub struct SpawnLock {
file: File,
}
impl SpawnLock {
pub fn acquire(paths: &DaemonPaths, wait: std::time::Duration) -> io::Result<Self> {
let path = paths.spawn_lock();
let file = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&path)?;
chmod_600(&path);
let deadline = std::time::Instant::now() + wait;
loop {
if FileExt::try_lock_exclusive(&file)? {
return Ok(Self { file });
}
if std::time::Instant::now() >= deadline {
return Err(io::Error::new(io::ErrorKind::TimedOut, "another client is still spawning the daemon"));
}
std::thread::sleep(std::time::Duration::from_millis(50));
}
}
}
impl Drop for SpawnLock {
fn drop(&mut self) {
let _ = FileExt::unlock(&self.file);
}
}
pub fn is_alive(paths: &DaemonPaths) -> bool {
if !paths.lock.exists() {
return false;
}
match DaemonLock::try_acquire(paths) {
Ok(Some(_guard)) => false,
Ok(None) => true,
Err(_) => false,
}
}
pub(super) fn cleanup_metadata(paths: &DaemonPaths, _lock: &DaemonLock) -> bool {
let mut removed = false;
for path in [&paths.json, &paths.pid] {
if std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_file()) {
match std::fs::remove_file(path) {
Ok(()) => removed = true,
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
Err(e) => tracing::warn!(path = %path.display(), error = %e, "daemon: metadata cleanup failed"),
}
}
}
removed
}
pub fn reap_stale(paths: &DaemonPaths) -> bool {
let Ok(Some(lock)) = DaemonLock::try_acquire(paths) else {
return false;
};
let had_socket = paths.sock.exists();
crate::events::socket::cleanup_socket(&paths.sock.to_string_lossy());
let reaped_socket = had_socket && !paths.sock.exists();
cleanup_metadata(paths, &lock) || reaped_socket
}
#[cfg(test)]
mod tests {
use super::*;
fn paths() -> (tempfile::TempDir, DaemonPaths) {
let d = tempfile::tempdir().unwrap();
let p = daemon_paths_in(d.path(), None);
(d, p)
}
#[test]
fn lock_is_the_liveness_oracle() {
let (_d, p) = paths();
assert!(!is_alive(&p));
let guard = DaemonLock::try_acquire(&p).unwrap().expect("first holder");
assert!(is_alive(&p));
assert!(DaemonLock::try_acquire(&p).unwrap().is_none(), "second holder refused");
drop(guard);
assert!(!is_alive(&p));
}
#[test]
fn spawn_lock_is_exclusive_and_waits() {
let (_d, p) = paths();
let a = SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
let err = SpawnLock::acquire(&p, std::time::Duration::from_millis(120)).unwrap_err();
assert_eq!(err.kind(), io::ErrorKind::TimedOut);
drop(a);
SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
assert!(!is_alive(&p), "the spawn lock is not the liveness oracle");
}
#[test]
fn metadata_cleanup_is_exact_and_requires_the_held_lock() {
let (d, p) = paths();
let other = daemon_paths_in(d.path(), Some("other"));
for path in [&p.json, &p.pid, &other.json, &other.pid] {
std::fs::write(path, b"metadata").unwrap();
}
let unrelated = d.path().join("session.json");
let unrelated_pid = d.path().join("session.pid");
std::fs::write(&unrelated, b"keep").unwrap();
std::fs::write(&unrelated_pid, b"keep").unwrap();
std::fs::write(&p.sock, b"not a socket").unwrap();
let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
assert!(cleanup_metadata(&p, &lock));
assert!(!p.json.exists() && !p.pid.exists());
assert!(is_alive(&p), "cleanup must not release or unlink the lock");
assert!(!cleanup_metadata(&p, &lock), "idempotent");
drop(lock);
assert!(!reap_stale(&p), "regular socket impostor is preserved");
for path in [&p.sock, &p.lock, &other.json, &other.pid, &unrelated, &unrelated_pid] {
assert!(path.exists(), "preserve {}", path.display());
}
}
#[cfg(unix)]
#[test]
fn metadata_cleanup_preserves_symlinks_and_special_files() {
use std::os::unix::{fs::symlink, net::UnixListener};
let (d, p) = paths();
let target = d.path().join("target");
std::fs::write(&target, b"keep").unwrap();
symlink(&target, &p.json).unwrap();
symlink(d.path().join("missing"), &p.pid).unwrap();
let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
assert!(!cleanup_metadata(&p, &lock));
drop(lock);
assert!(!reap_stale(&p));
for path in [&p.json, &p.pid] {
assert!(std::fs::symlink_metadata(path).unwrap().file_type().is_symlink());
std::fs::remove_file(path).unwrap();
}
assert_eq!(std::fs::read(&target).unwrap(), b"keep");
std::fs::create_dir(&p.json).unwrap();
let _listener = UnixListener::bind(&p.pid).unwrap();
assert!(!reap_stale(&p));
assert!(p.json.is_dir() && p.pid.exists());
}
fn info_for(p: &DaemonPaths, pid: u32) -> DaemonInfo {
DaemonInfo {
pid,
protocol_version: 3,
daemon_version: "1.2.3".into(),
profile: None,
started_at: chrono::Utc::now(),
socket: p.sock.to_string_lossy().into_owned(),
exe: Some("/opt/synaps".into()),
generation: 4,
}
}
#[test]
fn the_lock_holder_heals_a_missing_or_foreign_daemon_json() {
let (_d, p) = paths();
let ours = info_for(&p, 4_000_001);
let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
assert_eq!(read_daemon_json(&p), Some(ours.clone()));
assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000001\n");
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
assert_eq!(std::fs::metadata(&p.json).unwrap().permissions().mode() & 0o777, 0o600);
}
#[cfg(unix)]
let ino = {
use std::os::unix::fs::MetadataExt;
std::fs::metadata(&p.json).unwrap().ino()
};
assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
assert_eq!(std::fs::metadata(&p.json).unwrap().ino(), ino);
}
write_daemon_json(&p, &info_for(&p, 4_000_002)).unwrap();
assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
assert_eq!(read_daemon_json(&p), Some(ours.clone()));
std::fs::write(&p.json, b"not json").unwrap();
assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
assert_eq!(read_daemon_json(&p), Some(ours.clone()));
}
#[cfg(unix)]
#[test]
fn the_heal_leaves_symlinks_and_directories_alone() {
let (d, p) = paths();
let ours = info_for(&p, 4_000_003);
let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
let target = d.path().join("target");
std::fs::write(&target, b"keep").unwrap();
std::os::unix::fs::symlink(&target, &p.json).unwrap();
assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
assert!(std::fs::symlink_metadata(&p.json).unwrap().file_type().is_symlink());
assert_eq!(std::fs::read(&target).unwrap(), b"keep");
std::fs::remove_file(&p.json).unwrap();
std::fs::create_dir(&p.json).unwrap();
assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
assert!(p.json.is_dir());
}
fn rendezvous(n: usize) -> impl Fn() + Clone + Send + Sync + 'static {
let seen = std::sync::Arc::new((std::sync::Mutex::new(0usize), std::sync::Condvar::new()));
move || {
let (count, cv) = &*seen;
let mut c = count.lock().unwrap();
*c += 1;
cv.notify_all();
let _ = cv.wait_timeout_while(c, std::time::Duration::from_secs(5), |c| *c < n).unwrap();
}
}
#[test]
fn a_heal_paused_after_its_snapshot_never_writes_over_a_newer_record() {
let (_d, p) = paths();
let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
let old = info_for(&p, 4_000_010);
let meta = Metadata::start(p.clone(), old.clone()).unwrap();
std::fs::remove_file(&p.json).unwrap(); let new = DaemonInfo { generation: old.generation + 1, exe: Some("/opt/synaps-new".into()), ..old.clone() };
let (snapped_tx, snapped_rx) = std::sync::mpsc::channel::<()>();
let (published_tx, published_rx) = std::sync::mpsc::channel::<()>();
std::thread::scope(|s| {
let healer = s.spawn(|| {
test_pause::set(&test_pause::HEAL_BEFORE_WRITE, move || {
snapped_tx.send(()).unwrap();
let _ = published_rx.recv_timeout(std::time::Duration::from_millis(500));
});
meta.heal(&lock).unwrap()
});
snapped_rx.recv_timeout(std::time::Duration::from_secs(5)).expect("the heal is about to write");
meta.publish(new.clone()).unwrap();
let _ = published_tx.send(());
assert!(matches!(healer.join().unwrap(), Healed::Yes { .. }));
});
assert_eq!(read_daemon_json(&p), Some(new.clone()), "on disk: the newer record");
assert_eq!(meta.info(), new, "in memory: the newer record");
}
#[test]
fn daemon_json_and_daemon_pid_written_at_once_keep_their_own_contents() {
let (_d, p) = paths();
let meet = rendezvous(2);
let results: Vec<io::Result<()>> = std::thread::scope(|s| {
let a = s.spawn(|| {
test_pause::set(&test_pause::BEFORE_RENAME, meet.clone());
write_atomic(&p.json, b"{\"the\":\"json\"}")
});
let b = s.spawn(|| {
test_pause::set(&test_pause::BEFORE_RENAME, meet.clone());
write_atomic(&p.pid, b"4000011\n")
});
vec![a.join().unwrap(), b.join().unwrap()]
});
for r in &results {
assert!(r.is_ok(), "{results:?}");
}
assert_eq!(std::fs::read_to_string(&p.json).unwrap(), "{\"the\":\"json\"}");
assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000011\n");
}
#[cfg(unix)]
#[test]
fn a_symlink_at_a_temp_path_is_not_followed() {
use std::os::unix::fs::PermissionsExt;
let (d, p) = paths();
let target = d.path().join("target");
std::fs::write(&target, b"keep").unwrap();
for name in ["daemon.tmp", "daemon.json.tmp", "daemon.pid.tmp"] {
std::os::unix::fs::symlink(&target, d.path().join(name)).unwrap();
}
let ours = info_for(&p, 4_000_012);
write_daemon_json(&p, &ours).unwrap();
assert_eq!(std::fs::read(&target).unwrap(), b"keep");
assert_eq!(read_daemon_json(&p), Some(ours));
assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000012\n");
for path in [&p.json, &p.pid] {
assert!(std::fs::symlink_metadata(path).unwrap().file_type().is_file());
assert_eq!(std::fs::metadata(path).unwrap().permissions().mode() & 0o777, 0o600);
}
}
#[test]
fn nothing_writes_the_metadata_back_after_cleanup() {
let (_d, p) = paths();
let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
let ours = info_for(&p, 4_000_013);
let meta = Metadata::start(p.clone(), ours.clone()).unwrap();
assert!(meta.cleanup(&lock));
let _ = meta.publish(DaemonInfo { generation: 9, ..ours });
assert_eq!(meta.heal(&lock).unwrap(), Healed::No);
assert!(!p.json.exists() && !p.pid.exists());
}
#[test]
fn stale_files_reaped_only_without_holder() {
let (_d, p) = paths();
#[cfg(unix)]
let _listener = std::os::unix::net::UnixListener::bind(&p.sock).unwrap();
let info = DaemonInfo {
pid: 4_000_000,
protocol_version: 1,
daemon_version: "t".into(),
profile: None,
started_at: chrono::Utc::now(),
socket: p.sock.to_string_lossy().into_owned(),
exe: None,
generation: 1,
};
write_daemon_json(&p, &info).unwrap();
assert_eq!(read_daemon_json(&p).unwrap(), info);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
assert_eq!(std::fs::metadata(&p.json).unwrap().permissions().mode() & 0o777, 0o600);
}
let guard = DaemonLock::try_acquire(&p).unwrap().unwrap();
assert!(!reap_stale(&p), "live holder: nothing reaped");
#[cfg(unix)]
assert!(p.sock.exists());
assert!(p.json.exists() && p.pid.exists());
drop(guard);
assert!(reap_stale(&p));
assert!(!p.sock.exists() && !p.json.exists() && !p.pid.exists());
}
}