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 (tmp + rename), 0600.
47pub fn write_daemon_json(paths: &DaemonPaths, info: &DaemonInfo) -> io::Result<()> {
48    let body = serde_json::to_vec_pretty(info).map_err(io::Error::other)?;
49    write_atomic(&paths.json, &body)?;
50    write_atomic(&paths.pid, format!("{}\n", info.pid).as_bytes())
51}
52
53fn write_atomic(path: &Path, body: &[u8]) -> io::Result<()> {
54    let tmp = path.with_extension("tmp");
55    std::fs::write(&tmp, body)?;
56    chmod_600(&tmp);
57    std::fs::rename(&tmp, path)
58}
59
60/// 0600 tmp + rename (reload-state).
61pub fn write_private_atomic(path: &Path, body: &[u8]) -> io::Result<()> {
62    write_atomic(path, body)
63}
64
65pub fn read_daemon_json(paths: &DaemonPaths) -> Option<DaemonInfo> {
66    let s = std::fs::read_to_string(&paths.json).ok()?;
67    serde_json::from_str(&s).ok()
68}
69
70/// Held for the daemon's lifetime; dropping releases the flock.
71pub struct DaemonLock {
72    file: File,
73}
74
75impl DaemonLock {
76    /// `flock(LOCK_EX|LOCK_NB)` on `daemon.lock` (0600). `None` if another
77    /// daemon holds it.
78    pub fn try_acquire(paths: &DaemonPaths) -> io::Result<Option<Self>> {
79        let file = OpenOptions::new()
80            .create(true)
81            .truncate(false)
82            .read(true)
83            .write(true)
84            .open(&paths.lock)?;
85        chmod_600(&paths.lock);
86        if FileExt::try_lock_exclusive(&file)? {
87            Ok(Some(Self { file }))
88        } else {
89            Ok(None)
90        }
91    }
92}
93
94impl DaemonLock {
95    /// Adopt an already-locked fd inherited across `execv` (reload). The
96    /// flock travels with the open file description, so the new image
97    /// holds the SAME lock without a release/acquire gap.
98    #[cfg(unix)]
99    pub fn adopt(fd: i32) -> io::Result<Self> {
100        use std::os::unix::io::FromRawFd;
101        if fd < 0 {
102            return Err(io::Error::new(io::ErrorKind::InvalidInput, "negative lock fd"));
103        }
104        // SAFETY: the parent image cleared CLOEXEC on exactly this fd and
105        // handed it to us via SYNAPS_DAEMON_LOCK_FD; we take ownership.
106        let file = unsafe { File::from_raw_fd(fd) };
107        // Restore CLOEXEC so tool subprocesses never inherit the lock.
108        // SAFETY: fcntl on our own fd.
109        unsafe {
110            let flags = libc::fcntl(fd, libc::F_GETFD);
111            if flags >= 0 {
112                libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC);
113            }
114        }
115        Ok(Self { file })
116    }
117
118    #[cfg(unix)]
119    pub fn raw_fd(&self) -> i32 {
120        use std::os::unix::io::AsRawFd;
121        self.file.as_raw_fd()
122    }
123}
124
125impl Drop for DaemonLock {
126    fn drop(&mut self) {
127        let _ = FileExt::unlock(&self.file);
128    }
129}
130
131impl DaemonPaths {
132    /// `<stem>.spawn.lock` next to the flock: held by the one client that
133    /// is spawning the daemon (jcode's spawn lock). Never the liveness oracle.
134    pub fn spawn_lock(&self) -> std::path::PathBuf {
135        self.lock.with_extension("spawn.lock")
136    }
137}
138
139/// Held while a client spawns the daemon; dropping releases it.
140#[derive(Debug)]
141pub struct SpawnLock {
142    file: File,
143}
144
145impl SpawnLock {
146    /// `flock(LOCK_EX|LOCK_NB)` on the spawn lock, retried for up to `wait`
147    /// (a loser waits for the winner's spawn to finish, then re-probes the
148    /// daemon flock). `TimedOut` if the holder never lets go.
149    pub fn acquire(paths: &DaemonPaths, wait: std::time::Duration) -> io::Result<Self> {
150        let path = paths.spawn_lock();
151        let file = OpenOptions::new()
152            .create(true)
153            .truncate(false)
154            .read(true)
155            .write(true)
156            .open(&path)?;
157        chmod_600(&path);
158        let deadline = std::time::Instant::now() + wait;
159        loop {
160            if FileExt::try_lock_exclusive(&file)? {
161                return Ok(Self { file });
162            }
163            if std::time::Instant::now() >= deadline {
164                return Err(io::Error::new(io::ErrorKind::TimedOut, "another client is still spawning the daemon"));
165            }
166            std::thread::sleep(std::time::Duration::from_millis(50));
167        }
168    }
169}
170
171impl Drop for SpawnLock {
172    fn drop(&mut self) {
173        let _ = FileExt::unlock(&self.file);
174    }
175}
176
177/// Liveness probe: try to take the lock; success means nobody holds it
178/// (dead or never started) — release immediately. Never unlinks anything.
179pub fn is_alive(paths: &DaemonPaths) -> bool {
180    if !paths.lock.exists() {
181        return false;
182    }
183    match DaemonLock::try_acquire(paths) {
184        Ok(Some(_guard)) => false,
185        Ok(None) => true,
186        Err(_) => false,
187    }
188}
189
190/// Unlink `sock`/`json`/`pid` if — and only if — no daemon holds the lock.
191/// Returns `true` when something stale was removed.
192pub fn reap_stale(paths: &DaemonPaths) -> bool {
193    if is_alive(paths) {
194        return false;
195    }
196    let mut reaped = false;
197    for p in [&paths.sock, &paths.json, &paths.pid] {
198        if p.exists() {
199            crate::events::socket::cleanup_socket(&p.to_string_lossy());
200            reaped |= !p.exists();
201        }
202    }
203    reaped
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209
210    fn paths() -> (tempfile::TempDir, DaemonPaths) {
211        let d = tempfile::tempdir().unwrap();
212        let p = daemon_paths_in(d.path(), None);
213        (d, p)
214    }
215
216    #[test]
217    fn lock_is_the_liveness_oracle() {
218        let (_d, p) = paths();
219        assert!(!is_alive(&p));
220        let guard = DaemonLock::try_acquire(&p).unwrap().expect("first holder");
221        assert!(is_alive(&p));
222        assert!(DaemonLock::try_acquire(&p).unwrap().is_none(), "second holder refused");
223        drop(guard);
224        assert!(!is_alive(&p));
225    }
226
227    #[test]
228    fn spawn_lock_is_exclusive_and_waits() {
229        let (_d, p) = paths();
230        let a = SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
231        let err = SpawnLock::acquire(&p, std::time::Duration::from_millis(120)).unwrap_err();
232        assert_eq!(err.kind(), io::ErrorKind::TimedOut);
233        drop(a);
234        SpawnLock::acquire(&p, std::time::Duration::from_millis(10)).unwrap();
235        assert!(!is_alive(&p), "the spawn lock is not the liveness oracle");
236    }
237
238    #[test]
239    fn stale_files_reaped_only_without_holder() {
240        let (_d, p) = paths();
241        std::fs::write(&p.sock, b"").unwrap();
242        let info = DaemonInfo {
243            pid: 4_000_000,
244            protocol_version: 1,
245            daemon_version: "t".into(),
246            profile: None,
247            started_at: chrono::Utc::now(),
248            socket: p.sock.to_string_lossy().into_owned(),
249            exe: None,
250            generation: 1,
251        };
252        write_daemon_json(&p, &info).unwrap();
253        assert_eq!(read_daemon_json(&p).unwrap(), info);
254        #[cfg(unix)]
255        {
256            use std::os::unix::fs::PermissionsExt;
257            assert_eq!(std::fs::metadata(&p.json).unwrap().permissions().mode() & 0o777, 0o600);
258        }
259        let guard = DaemonLock::try_acquire(&p).unwrap().unwrap();
260        assert!(!reap_stale(&p), "live holder: nothing reaped");
261        assert!(p.sock.exists());
262        drop(guard);
263        assert!(reap_stale(&p));
264        assert!(!p.sock.exists() && !p.json.exists() && !p.pid.exists());
265    }
266}
agent_engine/daemon/registry.rs

agent_engine/daemon/
registry.rs
