1use std::fs::{File, OpenOptions};
8use std::io;
9use std::path::Path;
10
11use fs4::fs_std::FileExt;
12use serde::{Deserialize, Serialize};
13
14pub use crate::events::registry::{daemon_paths, daemon_paths_in, DaemonPaths};
15
16#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
18pub struct DaemonInfo {
19 pub pid: u32,
20 pub protocol_version: u32,
21 pub daemon_version: String,
22 pub profile: Option<String>,
23 pub started_at: chrono::DateTime<chrono::Utc>,
24 pub socket: String,
25 #[serde(default)]
28 pub exe: Option<std::path::PathBuf>,
29 #[serde(default = "one")]
31 pub generation: u64,
32}
33
34fn one() -> u64 {
35 1
36}
37
38#[cfg(unix)]
39fn chmod_600(path: &Path) {
40 use std::os::unix::fs::PermissionsExt;
41 let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600));
42}
43#[cfg(not(unix))]
44fn chmod_600(_path: &Path) {}
45
46pub fn write_daemon_json(paths: &DaemonPaths, info: &DaemonInfo) -> io::Result<()> {
49 let body = serde_json::to_vec_pretty(info).map_err(io::Error::other)?;
50 write_atomic(&paths.json, &body)?;
51 write_atomic(&paths.pid, format!("{}\n", info.pid).as_bytes())
52}
53
54fn temp_path(path: &Path) -> std::path::PathBuf {
57 let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
58 name.push(".tmp");
59 path.with_file_name(name)
60}
61
62fn create_private(path: &Path) -> io::Result<File> {
66 let mut options = OpenOptions::new();
67 options.write(true).create_new(true);
68 #[cfg(unix)]
69 {
70 use std::os::unix::fs::OpenOptionsExt;
71 options.mode(0o600).custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);
72 }
73 options.open(path)
74}
75
76#[cfg(unix)]
77fn set_private(file: &File) -> io::Result<()> {
78 use std::os::unix::fs::PermissionsExt;
79 file.set_permissions(std::fs::Permissions::from_mode(0o600))
81}
82#[cfg(not(unix))]
83fn set_private(_file: &File) -> io::Result<()> {
84 Ok(())
85}
86
87fn write_atomic(path: &Path, body: &[u8]) -> io::Result<()> {
91 use std::io::Write;
92 let tmp = temp_path(path);
93 let mut file = match create_private(&tmp) {
94 Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
95 std::fs::remove_file(&tmp)?;
96 create_private(&tmp)?
97 }
98 opened => opened?,
99 };
100 let written = file.write_all(body).and_then(|()| set_private(&file));
101 drop(file);
102 #[cfg(test)]
103 test_pause::at(&test_pause::BEFORE_RENAME);
104 let renamed = written.and_then(|()| std::fs::rename(&tmp, path));
105 if renamed.is_err() {
106 let _ = std::fs::remove_file(&tmp);
107 }
108 renamed
109}
110
111pub fn write_private_atomic(path: &Path, body: &[u8]) -> io::Result<()> {
113 write_atomic(path, body)
114}
115
116pub fn read_daemon_json(paths: &DaemonPaths) -> Option<DaemonInfo> {
117 let s = std::fs::read_to_string(&paths.json).ok()?;
118 serde_json::from_str(&s).ok()
119}
120
121pub fn heal_daemon_json(paths: &DaemonPaths, info: &DaemonInfo, _lock: &DaemonLock) -> io::Result<bool> {
129 match std::fs::symlink_metadata(&paths.json) {
130 Ok(m) if !m.file_type().is_file() => return Ok(false),
131 Ok(_) if read_daemon_json(paths).is_some_and(|on_disk| on_disk.pid == info.pid) => return Ok(false),
132 Ok(_) => {}
133 Err(e) if e.kind() == io::ErrorKind::NotFound => {}
134 Err(e) => return Err(e),
135 }
136 #[cfg(test)]
137 test_pause::at(&test_pause::HEAL_BEFORE_WRITE);
138 write_daemon_json(paths, info)?;
139 Ok(true)
140}
141
142pub const JSON_HEAL_LOG_EVERY: std::time::Duration = std::time::Duration::from_secs(600);
145
146pub struct Metadata {
157 paths: DaemonPaths,
158 inner: std::sync::Mutex<MetaInner>,
159}
160
161struct MetaInner {
162 info: DaemonInfo,
163 heals_logged_at: Option<std::time::Instant>,
166 heals_since_log: u64,
167 cleaned: bool,
169}
170
171#[derive(Debug, PartialEq, Eq)]
173pub enum Healed {
174 No,
177 Yes { info: DaemonInfo, log: Option<u64> },
180}
181
182impl Metadata {
183 pub fn start(paths: DaemonPaths, info: DaemonInfo) -> io::Result<Self> {
185 let meta = Self {
186 paths,
187 inner: std::sync::Mutex::new(MetaInner {
188 info,
189 heals_logged_at: None,
190 heals_since_log: 0,
191 cleaned: false,
192 }),
193 };
194 {
195 let inner = meta.inner();
196 write_daemon_json(&meta.paths, &inner.info)?;
197 }
198 Ok(meta)
199 }
200
201 fn inner(&self) -> std::sync::MutexGuard<'_, MetaInner> {
202 self.inner.lock().unwrap_or_else(|e| e.into_inner())
203 }
204
205 pub fn info(&self) -> DaemonInfo {
207 self.inner().info.clone()
208 }
209
210 pub fn publish(&self, info: DaemonInfo) -> io::Result<()> {
213 let mut inner = self.inner();
214 inner.info = info;
215 if inner.cleaned {
216 return Ok(());
217 }
218 write_daemon_json(&self.paths, &inner.info)
219 }
220
221 pub fn heal(&self, lock: &DaemonLock) -> io::Result<Healed> {
224 let mut inner = self.inner();
225 if inner.cleaned || !heal_daemon_json(&self.paths, &inner.info, lock)? {
226 return Ok(Healed::No);
227 }
228 inner.heals_since_log += 1;
229 let log = if inner.heals_logged_at.is_none_or(|at| at.elapsed() >= JSON_HEAL_LOG_EVERY) {
231 let heals = inner.heals_since_log;
232 inner.heals_logged_at = Some(std::time::Instant::now());
233 inner.heals_since_log = 0;
234 Some(heals)
235 } else {
236 None
237 };
238 Ok(Healed::Yes { info: inner.info.clone(), log })
239 }
240
241 pub fn cleanup(&self, lock: &DaemonLock) -> bool {
244 let mut inner = self.inner();
245 inner.cleaned = true;
246 cleanup_metadata(&self.paths, lock)
247 }
248}
249
250#[cfg(test)]
253pub(crate) mod test_pause {
254 use std::cell::RefCell;
255 use std::thread::LocalKey;
256
257 pub(crate) type Hook = RefCell<Option<Box<dyn Fn()>>>;
258
259 thread_local! {
260 pub(crate) static HEAL_BEFORE_WRITE: Hook = const { RefCell::new(None) };
263 pub(crate) static BEFORE_RENAME: Hook = const { RefCell::new(None) };
266 }
267
268 pub(crate) fn set(slot: &'static LocalKey<Hook>, hook: impl Fn() + 'static) {
269 slot.with(|h| *h.borrow_mut() = Some(Box::new(hook)));
270 }
271
272 pub(crate) fn at(slot: &'static LocalKey<Hook>) {
273 slot.with(|h| {
274 if let Some(hook) = h.borrow().as_ref() {
275 hook();
276 }
277 });
278 }
279}
280
281pub struct DaemonLock {
283 file: File,
284}
285
286impl DaemonLock {
287 pub fn try_acquire(paths: &DaemonPaths) -> io::Result<Option<Self>> {
290 let file = OpenOptions::new()
291 .create(true)
292 .truncate(false)
293 .read(true)
294 .write(true)
295 .open(&paths.lock)?;
296 chmod_600(&paths.lock);
297 if FileExt::try_lock_exclusive(&file)? {
298 Ok(Some(Self { file }))
299 } else {
300 Ok(None)
301 }
302 }
303}
304
305impl DaemonLock {
306 #[cfg(unix)]
310 pub fn adopt(fd: i32) -> io::Result<Self> {
311 use std::os::unix::io::FromRawFd;
312 if fd < 0 {
313 return Err(io::Error::new(io::ErrorKind::InvalidInput, "negative lock fd"));
314 }
315 let file = unsafe { File::from_raw_fd(fd) };
318 unsafe {
321 let flags = libc::fcntl(fd, libc::F_GETFD);
322 if flags >= 0 {
323 libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC);
324 }
325 }
326 Ok(Self { file })
327 }
328
329 #[cfg(unix)]
330 pub fn raw_fd(&self) -> i32 {
331 use std::os::unix::io::AsRawFd;
332 self.file.as_raw_fd()
333 }
334}
335
336impl Drop for DaemonLock {
337 fn drop(&mut self) {
338 let _ = FileExt::unlock(&self.file);
339 }
340}
341
342impl DaemonPaths {
343 pub fn spawn_lock(&self) -> std::path::PathBuf {
346 self.lock.with_extension("spawn.lock")
347 }
348}
349
350#[derive(Debug)]
352pub struct SpawnLock {
353 file: File,
354}
355
356impl SpawnLock {
357 pub fn acquire(paths: &DaemonPaths, wait: std::time::Duration) -> io::Result<Self> {
361 let path = paths.spawn_lock();
362 let file = OpenOptions::new()
363 .create(true)
364 .truncate(false)
365 .read(true)
366 .write(true)
367 .open(&path)?;
368 chmod_600(&path);
369 let deadline = std::time::Instant::now() + wait;
370 loop {
371 if FileExt::try_lock_exclusive(&file)? {
372 return Ok(Self { file });
373 }
374 if std::time::Instant::now() >= deadline {
375 return Err(io::Error::new(io::ErrorKind::TimedOut, "another client is still spawning the daemon"));
376 }
377 std::thread::sleep(std::time::Duration::from_millis(50));
378 }
379 }
380}
381
382impl Drop for SpawnLock {
383 fn drop(&mut self) {
384 let _ = FileExt::unlock(&self.file);
385 }
386}
387
388pub fn is_alive(paths: &DaemonPaths) -> bool {
391 if !paths.lock.exists() {
392 return false;
393 }
394 match DaemonLock::try_acquire(paths) {
395 Ok(Some(_guard)) => false,
396 Ok(None) => true,
397 Err(_) => false,
398 }
399}
400
401pub(super) fn cleanup_metadata(paths: &DaemonPaths, _lock: &DaemonLock) -> bool {
405 let mut removed = false;
406 for path in [&paths.json, &paths.pid] {
407 if std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_file()) {
408 match std::fs::remove_file(path) {
409 Ok(()) => removed = true,
410 Err(e) if e.kind() == io::ErrorKind::NotFound => {}
411 Err(e) => tracing::warn!(path = %path.display(), error = %e, "daemon: metadata cleanup failed"),
412 }
413 }
414 }
415 removed
416}
417
418pub fn reap_stale(paths: &DaemonPaths) -> bool {
421 let Ok(Some(lock)) = DaemonLock::try_acquire(paths) else {
423 return false;
424 };
425 let had_socket = paths.sock.exists();
426 crate::events::socket::cleanup_socket(&paths.sock.to_string_lossy());
427 let reaped_socket = had_socket && !paths.sock.exists();
428 cleanup_metadata(paths, &lock) || reaped_socket
429}
430
431#[cfg(test)]
432mod tests {
433 use super::*;
434
435 fn paths() -> (tempfile::TempDir, DaemonPaths) {
436 let d = tempfile::tempdir().unwrap();
437 let p = daemon_paths_in(d.path(), None);
438 (d, p)
439 }
440
441 #[test]
442 fn lock_is_the_liveness_oracle() {
443 let (_d, p) = paths();
444 assert!(!is_alive(&p));
445 let guard = DaemonLock::try_acquire(&p).unwrap().expect("first holder");
446 assert!(is_alive(&p));
447 assert!(DaemonLock::try_acquire(&p).unwrap().is_none(), "second holder refused");
448 drop(guard);
449 assert!(!is_alive(&p));
450 }
451
452 #[test]
453 fn spawn_lock_is_exclusive_and_waits() {
454 let (_d, p) = paths();
455 let a = SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
456 let err = SpawnLock::acquire(&p, std::time::Duration::from_millis(120)).unwrap_err();
457 assert_eq!(err.kind(), io::ErrorKind::TimedOut);
458 drop(a);
459 SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
460 assert!(!is_alive(&p), "the spawn lock is not the liveness oracle");
461 }
462
463 #[test]
464 fn metadata_cleanup_is_exact_and_requires_the_held_lock() {
465 let (d, p) = paths();
466 let other = daemon_paths_in(d.path(), Some("other"));
467 for path in [&p.json, &p.pid, &other.json, &other.pid] {
468 std::fs::write(path, b"metadata").unwrap();
469 }
470 let unrelated = d.path().join("session.json");
471 let unrelated_pid = d.path().join("session.pid");
472 std::fs::write(&unrelated, b"keep").unwrap();
473 std::fs::write(&unrelated_pid, b"keep").unwrap();
474 std::fs::write(&p.sock, b"not a socket").unwrap();
476 let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
477 assert!(cleanup_metadata(&p, &lock));
478 assert!(!p.json.exists() && !p.pid.exists());
479 assert!(is_alive(&p), "cleanup must not release or unlink the lock");
480 assert!(!cleanup_metadata(&p, &lock), "idempotent");
481 drop(lock);
482 assert!(!reap_stale(&p), "regular socket impostor is preserved");
483 for path in [&p.sock, &p.lock, &other.json, &other.pid, &unrelated, &unrelated_pid] {
484 assert!(path.exists(), "preserve {}", path.display());
485 }
486 }
487
488 #[cfg(unix)]
489 #[test]
490 fn metadata_cleanup_preserves_symlinks_and_special_files() {
491 use std::os::unix::{fs::symlink, net::UnixListener};
492 let (d, p) = paths();
493 let target = d.path().join("target");
494 std::fs::write(&target, b"keep").unwrap();
495 symlink(&target, &p.json).unwrap();
496 symlink(d.path().join("missing"), &p.pid).unwrap();
497 let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
498 assert!(!cleanup_metadata(&p, &lock));
499 drop(lock);
500 assert!(!reap_stale(&p));
501 for path in [&p.json, &p.pid] {
502 assert!(std::fs::symlink_metadata(path).unwrap().file_type().is_symlink());
503 std::fs::remove_file(path).unwrap();
504 }
505 assert_eq!(std::fs::read(&target).unwrap(), b"keep");
506 std::fs::create_dir(&p.json).unwrap();
507 let _listener = UnixListener::bind(&p.pid).unwrap();
508 assert!(!reap_stale(&p));
509 assert!(p.json.is_dir() && p.pid.exists());
510 }
511
512 fn info_for(p: &DaemonPaths, pid: u32) -> DaemonInfo {
513 DaemonInfo {
514 pid,
515 protocol_version: 3,
516 daemon_version: "1.2.3".into(),
517 profile: None,
518 started_at: chrono::Utc::now(),
519 socket: p.sock.to_string_lossy().into_owned(),
520 exe: Some("/opt/synaps".into()),
521 generation: 4,
522 }
523 }
524
525 #[test]
526 fn the_lock_holder_heals_a_missing_or_foreign_daemon_json() {
527 let (_d, p) = paths();
528 let ours = info_for(&p, 4_000_001);
529 let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
530 assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
532 assert_eq!(read_daemon_json(&p), Some(ours.clone()));
533 assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000001\n");
534 #[cfg(unix)]
535 {
536 use std::os::unix::fs::PermissionsExt;
537 assert_eq!(std::fs::metadata(&p.json).unwrap().permissions().mode() & 0o777, 0o600);
538 }
539 #[cfg(unix)]
541 let ino = {
542 use std::os::unix::fs::MetadataExt;
543 std::fs::metadata(&p.json).unwrap().ino()
544 };
545 assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
546 #[cfg(unix)]
547 {
548 use std::os::unix::fs::MetadataExt;
549 assert_eq!(std::fs::metadata(&p.json).unwrap().ino(), ino);
550 }
551 write_daemon_json(&p, &info_for(&p, 4_000_002)).unwrap();
553 assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
554 assert_eq!(read_daemon_json(&p), Some(ours.clone()));
555 std::fs::write(&p.json, b"not json").unwrap();
556 assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
557 assert_eq!(read_daemon_json(&p), Some(ours.clone()));
558 }
559
560 #[cfg(unix)]
561 #[test]
562 fn the_heal_leaves_symlinks_and_directories_alone() {
563 let (d, p) = paths();
564 let ours = info_for(&p, 4_000_003);
565 let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
566 let target = d.path().join("target");
567 std::fs::write(&target, b"keep").unwrap();
568 std::os::unix::fs::symlink(&target, &p.json).unwrap();
569 assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
570 assert!(std::fs::symlink_metadata(&p.json).unwrap().file_type().is_symlink());
571 assert_eq!(std::fs::read(&target).unwrap(), b"keep");
572 std::fs::remove_file(&p.json).unwrap();
573 std::fs::create_dir(&p.json).unwrap();
574 assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
575 assert!(p.json.is_dir());
576 }
577
578 fn rendezvous(n: usize) -> impl Fn() + Clone + Send + Sync + 'static {
580 let seen = std::sync::Arc::new((std::sync::Mutex::new(0usize), std::sync::Condvar::new()));
581 move || {
582 let (count, cv) = &*seen;
583 let mut c = count.lock().unwrap();
584 *c += 1;
585 cv.notify_all();
586 let _ = cv.wait_timeout_while(c, std::time::Duration::from_secs(5), |c| *c < n).unwrap();
587 }
588 }
589
590 #[test]
595 fn a_heal_paused_after_its_snapshot_never_writes_over_a_newer_record() {
596 let (_d, p) = paths();
597 let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
598 let old = info_for(&p, 4_000_010);
599 let meta = Metadata::start(p.clone(), old.clone()).unwrap();
600 std::fs::remove_file(&p.json).unwrap(); let new = DaemonInfo { generation: old.generation + 1, exe: Some("/opt/synaps-new".into()), ..old.clone() };
602 let (snapped_tx, snapped_rx) = std::sync::mpsc::channel::<()>();
603 let (published_tx, published_rx) = std::sync::mpsc::channel::<()>();
604 std::thread::scope(|s| {
605 let healer = s.spawn(|| {
606 test_pause::set(&test_pause::HEAL_BEFORE_WRITE, move || {
607 snapped_tx.send(()).unwrap();
608 let _ = published_rx.recv_timeout(std::time::Duration::from_millis(500));
611 });
612 meta.heal(&lock).unwrap()
613 });
614 snapped_rx.recv_timeout(std::time::Duration::from_secs(5)).expect("the heal is about to write");
615 meta.publish(new.clone()).unwrap();
616 let _ = published_tx.send(());
617 assert!(matches!(healer.join().unwrap(), Healed::Yes { .. }));
618 });
619 assert_eq!(read_daemon_json(&p), Some(new.clone()), "on disk: the newer record");
620 assert_eq!(meta.info(), new, "in memory: the newer record");
621 }
622
623 #[test]
627 fn daemon_json_and_daemon_pid_written_at_once_keep_their_own_contents() {
628 let (_d, p) = paths();
629 let meet = rendezvous(2);
630 let results: Vec<io::Result<()>> = std::thread::scope(|s| {
631 let a = s.spawn(|| {
632 test_pause::set(&test_pause::BEFORE_RENAME, meet.clone());
633 write_atomic(&p.json, b"{\"the\":\"json\"}")
634 });
635 let b = s.spawn(|| {
636 test_pause::set(&test_pause::BEFORE_RENAME, meet.clone());
637 write_atomic(&p.pid, b"4000011\n")
638 });
639 vec![a.join().unwrap(), b.join().unwrap()]
640 });
641 for r in &results {
642 assert!(r.is_ok(), "{results:?}");
643 }
644 assert_eq!(std::fs::read_to_string(&p.json).unwrap(), "{\"the\":\"json\"}");
645 assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000011\n");
646 }
647
648 #[cfg(unix)]
652 #[test]
653 fn a_symlink_at_a_temp_path_is_not_followed() {
654 use std::os::unix::fs::PermissionsExt;
655 let (d, p) = paths();
656 let target = d.path().join("target");
657 std::fs::write(&target, b"keep").unwrap();
658 for name in ["daemon.tmp", "daemon.json.tmp", "daemon.pid.tmp"] {
660 std::os::unix::fs::symlink(&target, d.path().join(name)).unwrap();
661 }
662 let ours = info_for(&p, 4_000_012);
663 write_daemon_json(&p, &ours).unwrap();
664 assert_eq!(std::fs::read(&target).unwrap(), b"keep");
665 assert_eq!(read_daemon_json(&p), Some(ours));
666 assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000012\n");
667 for path in [&p.json, &p.pid] {
668 assert!(std::fs::symlink_metadata(path).unwrap().file_type().is_file());
669 assert_eq!(std::fs::metadata(path).unwrap().permissions().mode() & 0o777, 0o600);
670 }
671 }
672
673 #[test]
676 fn nothing_writes_the_metadata_back_after_cleanup() {
677 let (_d, p) = paths();
678 let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
679 let ours = info_for(&p, 4_000_013);
680 let meta = Metadata::start(p.clone(), ours.clone()).unwrap();
681 assert!(meta.cleanup(&lock));
682 let _ = meta.publish(DaemonInfo { generation: 9, ..ours });
683 assert_eq!(meta.heal(&lock).unwrap(), Healed::No);
684 assert!(!p.json.exists() && !p.pid.exists());
685 }
686
687 #[test]
688 fn stale_files_reaped_only_without_holder() {
689 let (_d, p) = paths();
690 #[cfg(unix)]
691 let _listener = std::os::unix::net::UnixListener::bind(&p.sock).unwrap();
692 let info = DaemonInfo {
693 pid: 4_000_000,
694 protocol_version: 1,
695 daemon_version: "t".into(),
696 profile: None,
697 started_at: chrono::Utc::now(),
698 socket: p.sock.to_string_lossy().into_owned(),
699 exe: None,
700 generation: 1,
701 };
702 write_daemon_json(&p, &info).unwrap();
703 assert_eq!(read_daemon_json(&p).unwrap(), info);
704 #[cfg(unix)]
705 {
706 use std::os::unix::fs::PermissionsExt;
707 assert_eq!(std::fs::metadata(&p.json).unwrap().permissions().mode() & 0o777, 0o600);
708 }
709 let guard = DaemonLock::try_acquire(&p).unwrap().unwrap();
710 assert!(!reap_stale(&p), "live holder: nothing reaped");
711 #[cfg(unix)]
712 assert!(p.sock.exists());
713 assert!(p.json.exists() && p.pid.exists());
714 drop(guard);
715 assert!(reap_stale(&p));
716 assert!(!p.sock.exists() && !p.json.exists() && !p.pid.exists());
717 }
718}