Skip to main content

agent_engine/daemon/
registry.rs

1//! `daemon.json` / `daemon.lock` / `daemon.pid` under `registry_dir()`
2//! (PLAN-phase2 §2.11). The flock on `daemon.lock` is THE liveness oracle
3//! (jcode `socket.rs:89-139`): a daemon is alive iff someone holds the lock.
4//! A socket file with no lock holder is stale and gets unlinked. Nobody
5//! unlinks on ECONNREFUSED alone.
6
7use 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/// `daemon.json` — pid + profile + paths only. Never credentials.
17#[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    /// Executable to re-exec on `reload` (argv[0] canonicalised at first
26    /// start; `--exe` overrides and is recorded). C3.
27    #[serde(default)]
28    pub exe: Option<std::path::PathBuf>,
29    /// Reload counter (starts at 1; `Welcome.generation`). C3.
30    #[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
46/// Write `daemon.json` + `daemon.pid` atomically (each through its own
47/// temp file + rename), 0600.
48pub 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
54/// `<file name>.tmp` next to `path`: one temp file per target, so writes of
55/// `daemon.json` and `daemon.pid` never share one.
56fn 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
62/// Create `path` as a new private file: `O_CREAT | O_EXCL` (never opens an
63/// existing file, and so never a symlink), `O_NOFOLLOW`, mode 0600 at
64/// creation.
65fn 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    // Exactly 0600, whatever the umask took away at creation.
80    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
87/// Write `body` to `<path>.tmp`, created exclusively (a leftover or a
88/// planted file there is unlinked first, never followed), 0600, then rename
89/// it over `path`. The temp file is removed when anything fails.
90fn 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
111/// 0600 temp file + rename (reload-state), as [`write_atomic`].
112pub 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
121/// The lock holder puts its own `daemon.json` (and `daemon.pid`) back when
122/// the file is missing, unreadable, or names another pid: an older `synaps`
123/// that lists sessions unlinks every run-dir `*.json` it cannot parse as a
124/// session card. `info` is the record the daemon holds in memory; the write
125/// is [`write_daemon_json`]'s (tmp + rename, 0600). Anything at the path
126/// that is not a regular file (a symlink, a directory, a socket) is left
127/// alone, as [`cleanup_metadata`] does. Returns whether it wrote.
128pub 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
142/// At most one `daemon.json rewritten` line per this interval; heals in
143/// between are counted in the next line.
144pub const JSON_HEAL_LOG_EVERY: std::time::Duration = std::time::Duration::from_secs(600);
145
146/// The daemon's `daemon.json` record as this image holds it, and the writes
147/// of it: the start write, [`Self::publish`] (reload, and the restore after
148/// a failed exec), [`Self::heal`] (Ping) and [`Self::cleanup`] (shutdown).
149///
150/// One metadata lock serializes them all: each changes the held record and
151/// writes the files inside the same critical section, so a heal can never
152/// write a record older than the one held, and after the cleanup nothing
153/// writes the files back. Lock order: a caller that also holds the daemon's
154/// `lock_fd` mutex (heal, cleanup) takes that first. The lock is never held
155/// across an await.
156pub struct Metadata {
157    paths: DaemonPaths,
158    inner: std::sync::Mutex<MetaInner>,
159}
160
161struct MetaInner {
162    info: DaemonInfo,
163    /// The heal's INFO-line rate limit: when it was last logged, and the
164    /// heals since then.
165    heals_logged_at: Option<std::time::Instant>,
166    heals_since_log: u64,
167    /// Shutdown removed the files: no later write puts them back.
168    cleaned: bool,
169}
170
171/// What [`Metadata::heal`] did.
172#[derive(Debug, PartialEq, Eq)]
173pub enum Healed {
174    /// The file was in place (or is not a regular file), or the metadata
175    /// was cleaned up: nothing written.
176    No,
177    /// The record was written back. `log` is `Some(heals)` when this heal
178    /// is due its INFO line (`heals` counted since the last line).
179    Yes { info: DaemonInfo, log: Option<u64> },
180}
181
182impl Metadata {
183    /// Hold `info` and write it (the start write).
184    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    /// The record this image holds (a copy).
206    pub fn info(&self) -> DaemonInfo {
207        self.inner().info.clone()
208    }
209
210    /// Hold `info` and write it, under the metadata lock. After the cleanup
211    /// it is only held.
212    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    /// [`heal_daemon_json`] with the held record, under the metadata lock;
222    /// counts the heal for the INFO line's rate limit.
223    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        // `map_or`, not `is_none_or`: the crate's MSRV is 1.80.
230        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    /// [`cleanup_metadata`] (shutdown), under the metadata lock; no later
242    /// publish or heal writes the files.
243    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/// Test-only pause points (thread-local, so a test pauses only its own
251/// thread): a hook set here runs at that point of the write path.
252#[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        /// In [`super::heal_daemon_json`], after it found the file missing
261        /// or foreign and before it writes the record it was given.
262        pub(crate) static HEAL_BEFORE_WRITE: Hook = const { RefCell::new(None) };
263        /// In `write_atomic`, after the temp file is written and before it
264        /// is renamed into place.
265        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
281/// Held for the daemon's lifetime; dropping releases the flock.
282pub struct DaemonLock {
283    file: File,
284}
285
286impl DaemonLock {
287    /// `flock(LOCK_EX|LOCK_NB)` on `daemon.lock` (0600). `None` if another
288    /// daemon holds it.
289    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    /// Adopt an already-locked fd inherited across `execv` (reload). The
307    /// flock travels with the open file description, so the new image
308    /// holds the SAME lock without a release/acquire gap.
309    #[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        // SAFETY: the parent image cleared CLOEXEC on exactly this fd and
316        // handed it to us via SYNAPS_DAEMON_LOCK_FD; we take ownership.
317        let file = unsafe { File::from_raw_fd(fd) };
318        // Restore CLOEXEC so tool subprocesses never inherit the lock.
319        // SAFETY: fcntl on our own fd.
320        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
336#[cfg(windows)]
337impl DaemonLock {
338    /// Reload hand-off — Windows has no `execv`, and a byte-range lock
339    /// belongs to the process that took it (see [`super::win`]). Queue an
340    /// exclusive lock on `daemon.lock` behind the previous image, run
341    /// `queued` once the request sits in the kernel's queue (the previous
342    /// image exits on that signal), then wait up to `wait` for the grant.
343    /// No prober can take the lock in between: the release grants it here.
344    pub fn acquire_handoff(
345        paths: &DaemonPaths,
346        wait: std::time::Duration,
347        queued: impl FnOnce(),
348    ) -> io::Result<Self> {
349        use std::os::windows::fs::OpenOptionsExt;
350        let file = OpenOptions::new()
351            .create(true)
352            .truncate(false)
353            .read(true)
354            .write(true)
355            // Overlapped, so the lock request queues instead of blocking.
356            .custom_flags(windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OVERLAPPED)
357            .open(&paths.lock)?;
358        super::win::lock_exclusive_queued(&file, wait, queued)?;
359        Ok(Self { file })
360    }
361}
362
363impl Drop for DaemonLock {
364    fn drop(&mut self) {
365        let _ = FileExt::unlock(&self.file);
366    }
367}
368
369impl DaemonPaths {
370    /// `<stem>.spawn.lock` next to the flock: held by the one client that
371    /// is spawning the daemon (jcode's spawn lock). Never the liveness oracle.
372    pub fn spawn_lock(&self) -> std::path::PathBuf {
373        self.lock.with_extension("spawn.lock")
374    }
375}
376
377/// Held while a client spawns the daemon; dropping releases it.
378#[derive(Debug)]
379pub struct SpawnLock {
380    file: File,
381}
382
383impl SpawnLock {
384    /// `flock(LOCK_EX|LOCK_NB)` on the spawn lock, retried for up to `wait`
385    /// (a loser waits for the winner's spawn to finish, then re-probes the
386    /// daemon flock). `TimedOut` if the holder never lets go.
387    pub fn acquire(paths: &DaemonPaths, wait: std::time::Duration) -> io::Result<Self> {
388        let path = paths.spawn_lock();
389        let file = OpenOptions::new()
390            .create(true)
391            .truncate(false)
392            .read(true)
393            .write(true)
394            .open(&path)?;
395        chmod_600(&path);
396        let deadline = std::time::Instant::now() + wait;
397        loop {
398            if FileExt::try_lock_exclusive(&file)? {
399                return Ok(Self { file });
400            }
401            if std::time::Instant::now() >= deadline {
402                return Err(io::Error::new(io::ErrorKind::TimedOut, "another client is still spawning the daemon"));
403            }
404            std::thread::sleep(std::time::Duration::from_millis(50));
405        }
406    }
407}
408
409impl Drop for SpawnLock {
410    fn drop(&mut self) {
411        let _ = FileExt::unlock(&self.file);
412    }
413}
414
415/// Liveness probe: try to take the lock; success means nobody holds it
416/// (dead or never started) — release immediately. Never unlinks anything.
417pub fn is_alive(paths: &DaemonPaths) -> bool {
418    if !paths.lock.exists() {
419        return false;
420    }
421    match DaemonLock::try_acquire(paths) {
422        Ok(Some(_guard)) => false,
423        Ok(None) => true,
424        Err(_) => false,
425    }
426}
427
428/// Remove only the daemon's two metadata files, while its lock is held.
429/// These paths come from DaemonPaths, never from a session card or a suffix scan.
430/// Refuse symlinks and special files (including directories and sockets).
431pub(super) fn cleanup_metadata(paths: &DaemonPaths, _lock: &DaemonLock) -> bool {
432    let mut removed = false;
433    for path in [&paths.json, &paths.pid] {
434        if std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_file()) {
435            match std::fs::remove_file(path) {
436                Ok(()) => removed = true,
437                Err(e) if e.kind() == io::ErrorKind::NotFound => {}
438                Err(e) => tracing::warn!(path = %path.display(), error = %e, "daemon: metadata cleanup failed"),
439            }
440        }
441    }
442    removed
443}
444
445/// Unlink `sock`/`json`/`pid` if — and only if — no daemon holds the lock.
446/// Returns `true` when something stale was removed.
447pub fn reap_stale(paths: &DaemonPaths) -> bool {
448    // Retain the probe's lock through deletion: metadata cleanup requires ownership.
449    let Ok(Some(lock)) = DaemonLock::try_acquire(paths) else {
450        return false;
451    };
452    let had_socket = paths.sock.exists();
453    crate::events::socket::cleanup_socket(&paths.sock.to_string_lossy());
454    let reaped_socket = had_socket && !paths.sock.exists();
455    cleanup_metadata(paths, &lock) || reaped_socket
456}
457
458#[cfg(test)]
459mod tests {
460    use super::*;
461
462    fn paths() -> (tempfile::TempDir, DaemonPaths) {
463        let d = tempfile::tempdir().unwrap();
464        let p = daemon_paths_in(d.path(), None);
465        (d, p)
466    }
467
468    #[test]
469    fn lock_is_the_liveness_oracle() {
470        let (_d, p) = paths();
471        assert!(!is_alive(&p));
472        let guard = DaemonLock::try_acquire(&p).unwrap().expect("first holder");
473        assert!(is_alive(&p));
474        assert!(DaemonLock::try_acquire(&p).unwrap().is_none(), "second holder refused");
475        drop(guard);
476        assert!(!is_alive(&p));
477    }
478
479    #[test]
480    fn spawn_lock_is_exclusive_and_waits() {
481        let (_d, p) = paths();
482        let a = SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
483        let err = SpawnLock::acquire(&p, std::time::Duration::from_millis(120)).unwrap_err();
484        assert_eq!(err.kind(), io::ErrorKind::TimedOut);
485        drop(a);
486        SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
487        assert!(!is_alive(&p), "the spawn lock is not the liveness oracle");
488    }
489
490    #[test]
491    fn metadata_cleanup_is_exact_and_requires_the_held_lock() {
492        let (d, p) = paths();
493        let other = daemon_paths_in(d.path(), Some("other"));
494        for path in [&p.json, &p.pid, &other.json, &other.pid] {
495            std::fs::write(path, b"metadata").unwrap();
496        }
497        let unrelated = d.path().join("session.json");
498        let unrelated_pid = d.path().join("session.pid");
499        std::fs::write(&unrelated, b"keep").unwrap();
500        std::fs::write(&unrelated_pid, b"keep").unwrap();
501        // A regular file at the socket path must not be removed either.
502        std::fs::write(&p.sock, b"not a socket").unwrap();
503        let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
504        assert!(cleanup_metadata(&p, &lock));
505        assert!(!p.json.exists() && !p.pid.exists());
506        assert!(is_alive(&p), "cleanup must not release or unlink the lock");
507        assert!(!cleanup_metadata(&p, &lock), "idempotent");
508        drop(lock);
509        assert!(!reap_stale(&p), "regular socket impostor is preserved");
510        for path in [&p.sock, &p.lock, &other.json, &other.pid, &unrelated, &unrelated_pid] {
511            assert!(path.exists(), "preserve {}", path.display());
512        }
513    }
514
515    #[cfg(unix)]
516    #[test]
517    fn metadata_cleanup_preserves_symlinks_and_special_files() {
518        use std::os::unix::{fs::symlink, net::UnixListener};
519        let (d, p) = paths();
520        let target = d.path().join("target");
521        std::fs::write(&target, b"keep").unwrap();
522        symlink(&target, &p.json).unwrap();
523        symlink(d.path().join("missing"), &p.pid).unwrap();
524        let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
525        assert!(!cleanup_metadata(&p, &lock));
526        drop(lock);
527        assert!(!reap_stale(&p));
528        for path in [&p.json, &p.pid] {
529            assert!(std::fs::symlink_metadata(path).unwrap().file_type().is_symlink());
530            std::fs::remove_file(path).unwrap();
531        }
532        assert_eq!(std::fs::read(&target).unwrap(), b"keep");
533        std::fs::create_dir(&p.json).unwrap();
534        let _listener = UnixListener::bind(&p.pid).unwrap();
535        assert!(!reap_stale(&p));
536        assert!(p.json.is_dir() && p.pid.exists());
537    }
538
539    fn info_for(p: &DaemonPaths, pid: u32) -> DaemonInfo {
540        DaemonInfo {
541            pid,
542            protocol_version: 3,
543            daemon_version: "1.2.3".into(),
544            profile: None,
545            started_at: chrono::Utc::now(),
546            socket: p.sock.to_string_lossy().into_owned(),
547            exe: Some("/opt/synaps".into()),
548            generation: 4,
549        }
550    }
551
552    #[test]
553    fn the_lock_holder_heals_a_missing_or_foreign_daemon_json() {
554        let (_d, p) = paths();
555        let ours = info_for(&p, 4_000_001);
556        let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
557        // Missing: written, with daemon.pid, 0600.
558        assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
559        assert_eq!(read_daemon_json(&p), Some(ours.clone()));
560        assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000001\n");
561        #[cfg(unix)]
562        {
563            use std::os::unix::fs::PermissionsExt;
564            assert_eq!(std::fs::metadata(&p.json).unwrap().permissions().mode() & 0o777, 0o600);
565        }
566        // Ours already: no write (same inode).
567        #[cfg(unix)]
568        let ino = {
569            use std::os::unix::fs::MetadataExt;
570            std::fs::metadata(&p.json).unwrap().ino()
571        };
572        assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
573        #[cfg(unix)]
574        {
575            use std::os::unix::fs::MetadataExt;
576            assert_eq!(std::fs::metadata(&p.json).unwrap().ino(), ino);
577        }
578        // Another pid, or not a DaemonInfo: replaced.
579        write_daemon_json(&p, &info_for(&p, 4_000_002)).unwrap();
580        assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
581        assert_eq!(read_daemon_json(&p), Some(ours.clone()));
582        std::fs::write(&p.json, b"not json").unwrap();
583        assert!(heal_daemon_json(&p, &ours, &lock).unwrap());
584        assert_eq!(read_daemon_json(&p), Some(ours.clone()));
585    }
586
587    #[cfg(unix)]
588    #[test]
589    fn the_heal_leaves_symlinks_and_directories_alone() {
590        let (d, p) = paths();
591        let ours = info_for(&p, 4_000_003);
592        let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
593        let target = d.path().join("target");
594        std::fs::write(&target, b"keep").unwrap();
595        std::os::unix::fs::symlink(&target, &p.json).unwrap();
596        assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
597        assert!(std::fs::symlink_metadata(&p.json).unwrap().file_type().is_symlink());
598        assert_eq!(std::fs::read(&target).unwrap(), b"keep");
599        std::fs::remove_file(&p.json).unwrap();
600        std::fs::create_dir(&p.json).unwrap();
601        assert!(!heal_daemon_json(&p, &ours, &lock).unwrap());
602        assert!(p.json.is_dir());
603    }
604
605    /// Wait (at most 5 s) until `n` threads have arrived here.
606    fn rendezvous(n: usize) -> impl Fn() + Clone + Send + Sync + 'static {
607        let seen = std::sync::Arc::new((std::sync::Mutex::new(0usize), std::sync::Condvar::new()));
608        move || {
609            let (count, cv) = &*seen;
610            let mut c = count.lock().unwrap();
611            *c += 1;
612            cv.notify_all();
613            let _ = cv.wait_timeout_while(c, std::time::Duration::from_secs(5), |c| *c < n).unwrap();
614        }
615    }
616
617    /// A heal that took the held record, found the file missing and then
618    /// stalls before its write (a pause point) while a reload publishes a newer
619    /// record must not write its older copy over the newer one: the newer
620    /// record ends on disk and in memory.
621    #[test]
622    fn a_heal_paused_after_its_snapshot_never_writes_over_a_newer_record() {
623        let (_d, p) = paths();
624        let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
625        let old = info_for(&p, 4_000_010);
626        let meta = Metadata::start(p.clone(), old.clone()).unwrap();
627        std::fs::remove_file(&p.json).unwrap(); // so the heal writes
628        let new = DaemonInfo { generation: old.generation + 1, exe: Some("/opt/synaps-new".into()), ..old.clone() };
629        let (snapped_tx, snapped_rx) = std::sync::mpsc::channel::<()>();
630        let (published_tx, published_rx) = std::sync::mpsc::channel::<()>();
631        std::thread::scope(|s| {
632            let healer = s.spawn(|| {
633                test_pause::set(&test_pause::HEAL_BEFORE_WRITE, move || {
634                    snapped_tx.send(()).unwrap();
635                    // Held here until the publish is done, or 500 ms if the
636                    // publish waits for this heal.
637                    let _ = published_rx.recv_timeout(std::time::Duration::from_millis(500));
638                });
639                meta.heal(&lock).unwrap()
640            });
641            snapped_rx.recv_timeout(std::time::Duration::from_secs(5)).expect("the heal is about to write");
642            meta.publish(new.clone()).unwrap();
643            let _ = published_tx.send(());
644            assert!(matches!(healer.join().unwrap(), Healed::Yes { .. }));
645        });
646        assert_eq!(read_daemon_json(&p), Some(new.clone()), "on disk: the newer record");
647        assert_eq!(meta.info(), new, "in memory: the newer record");
648    }
649
650    /// `daemon.json` and `daemon.pid` written at the same moment (both temp
651    /// files written before either is renamed) each end with their own
652    /// contents: the two writes never share a temp file.
653    #[test]
654    fn daemon_json_and_daemon_pid_written_at_once_keep_their_own_contents() {
655        let (_d, p) = paths();
656        let meet = rendezvous(2);
657        let results: Vec<io::Result<()>> = std::thread::scope(|s| {
658            let a = s.spawn(|| {
659                test_pause::set(&test_pause::BEFORE_RENAME, meet.clone());
660                write_atomic(&p.json, b"{\"the\":\"json\"}")
661            });
662            let b = s.spawn(|| {
663                test_pause::set(&test_pause::BEFORE_RENAME, meet.clone());
664                write_atomic(&p.pid, b"4000011\n")
665            });
666            vec![a.join().unwrap(), b.join().unwrap()]
667        });
668        for r in &results {
669            assert!(r.is_ok(), "{results:?}");
670        }
671        assert_eq!(std::fs::read_to_string(&p.json).unwrap(), "{\"the\":\"json\"}");
672        assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000011\n");
673    }
674
675    /// A symlink planted where a temp file goes is never followed: the
676    /// write replaces the link, the file it points at keeps its contents,
677    /// and the written files are 0600.
678    #[cfg(unix)]
679    #[test]
680    fn a_symlink_at_a_temp_path_is_not_followed() {
681        use std::os::unix::fs::PermissionsExt;
682        let (d, p) = paths();
683        let target = d.path().join("target");
684        std::fs::write(&target, b"keep").unwrap();
685        // The shared temp name of earlier versions and the per-file names.
686        for name in ["daemon.tmp", "daemon.json.tmp", "daemon.pid.tmp"] {
687            std::os::unix::fs::symlink(&target, d.path().join(name)).unwrap();
688        }
689        let ours = info_for(&p, 4_000_012);
690        write_daemon_json(&p, &ours).unwrap();
691        assert_eq!(std::fs::read(&target).unwrap(), b"keep");
692        assert_eq!(read_daemon_json(&p), Some(ours));
693        assert_eq!(std::fs::read_to_string(&p.pid).unwrap(), "4000012\n");
694        for path in [&p.json, &p.pid] {
695            assert!(std::fs::symlink_metadata(path).unwrap().file_type().is_file());
696            assert_eq!(std::fs::metadata(path).unwrap().permissions().mode() & 0o777, 0o600);
697        }
698    }
699
700    /// After shutdown's cleanup no later write puts the files back: neither
701    /// a publish (a reload racing the shutdown) nor a heal.
702    #[test]
703    fn nothing_writes_the_metadata_back_after_cleanup() {
704        let (_d, p) = paths();
705        let lock = DaemonLock::try_acquire(&p).unwrap().unwrap();
706        let ours = info_for(&p, 4_000_013);
707        let meta = Metadata::start(p.clone(), ours.clone()).unwrap();
708        assert!(meta.cleanup(&lock));
709        let _ = meta.publish(DaemonInfo { generation: 9, ..ours });
710        assert_eq!(meta.heal(&lock).unwrap(), Healed::No);
711        assert!(!p.json.exists() && !p.pid.exists());
712    }
713
714    #[test]
715    fn stale_files_reaped_only_without_holder() {
716        let (_d, p) = paths();
717        #[cfg(unix)]
718        let _listener = std::os::unix::net::UnixListener::bind(&p.sock).unwrap();
719        let info = DaemonInfo {
720            pid: 4_000_000,
721            protocol_version: 1,
722            daemon_version: "t".into(),
723            profile: None,
724            started_at: chrono::Utc::now(),
725            socket: p.sock.to_string_lossy().into_owned(),
726            exe: None,
727            generation: 1,
728        };
729        write_daemon_json(&p, &info).unwrap();
730        assert_eq!(read_daemon_json(&p).unwrap(), info);
731        #[cfg(unix)]
732        {
733            use std::os::unix::fs::PermissionsExt;
734            assert_eq!(std::fs::metadata(&p.json).unwrap().permissions().mode() & 0o777, 0o600);
735        }
736        let guard = DaemonLock::try_acquire(&p).unwrap().unwrap();
737        assert!(!reap_stale(&p), "live holder: nothing reaped");
738        #[cfg(unix)]
739        assert!(p.sock.exists());
740        assert!(p.json.exists() && p.pid.exists());
741        drop(guard);
742        assert!(reap_stale(&p));
743        assert!(!p.sock.exists() && !p.json.exists() && !p.pid.exists());
744    }
745}