zeph_core/
daemon.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Daemon supervisor for component lifecycle management.
5
6use std::time::Duration;
7
8use tokio::sync::watch;
9use tokio::task::JoinHandle;
10
11use crate::config::DaemonConfig;
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub enum ComponentStatus {
15    Running,
16    Failed(String),
17    Stopped,
18}
19
20/// Error type for daemon component task failures.
21#[derive(Debug, thiserror::Error)]
22pub enum DaemonError {
23    #[error("task error: {0}")]
24    Task(String),
25    #[error("shutdown error: {0}")]
26    Shutdown(String),
27}
28
29pub struct ComponentHandle {
30    pub name: String,
31    handle: JoinHandle<Result<(), DaemonError>>,
32    pub status: ComponentStatus,
33    pub restart_count: u32,
34}
35
36impl ComponentHandle {
37    #[must_use]
38    pub fn new(name: impl Into<String>, handle: JoinHandle<Result<(), DaemonError>>) -> Self {
39        Self {
40            name: name.into(),
41            handle,
42            status: ComponentStatus::Running,
43            restart_count: 0,
44        }
45    }
46
47    #[must_use]
48    pub fn is_finished(&self) -> bool {
49        self.handle.is_finished()
50    }
51}
52
53pub struct DaemonSupervisor {
54    components: Vec<ComponentHandle>,
55    health_interval: Duration,
56    _max_backoff: Duration,
57    shutdown_rx: watch::Receiver<bool>,
58}
59
60impl DaemonSupervisor {
61    #[must_use]
62    pub fn new(config: &DaemonConfig, shutdown_rx: watch::Receiver<bool>) -> Self {
63        Self {
64            components: Vec::new(),
65            health_interval: Duration::from_secs(config.health_interval_secs),
66            _max_backoff: Duration::from_secs(config.max_restart_backoff_secs),
67            shutdown_rx,
68        }
69    }
70
71    pub fn add_component(&mut self, handle: ComponentHandle) {
72        self.components.push(handle);
73    }
74
75    #[must_use]
76    pub fn component_count(&self) -> usize {
77        self.components.len()
78    }
79
80    /// Run the health monitoring loop until shutdown signal.
81    pub async fn run(&mut self) {
82        let mut interval = tokio::time::interval(self.health_interval);
83        loop {
84            tokio::select! {
85                _ = interval.tick() => {
86                    self.check_health();
87                }
88                _ = self.shutdown_rx.changed() => {
89                    if *self.shutdown_rx.borrow() {
90                        tracing::info!("daemon supervisor shutting down");
91                        break;
92                    }
93                }
94            }
95        }
96    }
97
98    fn check_health(&mut self) {
99        for component in &mut self.components {
100            if component.status == ComponentStatus::Running && component.is_finished() {
101                component.status = ComponentStatus::Failed("task exited".into());
102                component.restart_count += 1;
103                tracing::warn!(
104                    component = %component.name,
105                    restarts = component.restart_count,
106                    "component exited unexpectedly"
107                );
108            }
109        }
110    }
111
112    #[must_use]
113    pub fn component_statuses(&self) -> Vec<(&str, &ComponentStatus)> {
114        self.components
115            .iter()
116            .map(|c| (c.name.as_str(), &c.status))
117            .collect()
118    }
119}
120
121/// Write a PID file atomically using `O_CREAT | O_EXCL` to prevent TOCTOU races.
122///
123/// # Errors
124///
125/// Returns an error if the PID file directory cannot be created, the file already exists,
126/// or the file cannot be written.
127pub fn write_pid_file(path: &str) -> std::io::Result<()> {
128    use std::io::Write as _;
129    let expanded = expand_tilde(path);
130    let path = std::path::Path::new(&expanded);
131    if let Some(parent) = path.parent() {
132        std::fs::create_dir_all(parent)?;
133    }
134    let mut file = std::fs::OpenOptions::new()
135        .write(true)
136        .create_new(true)
137        .open(path)?;
138    file.write_all(std::process::id().to_string().as_bytes())
139}
140
141/// Read the PID from a PID file.
142///
143/// # Errors
144///
145/// Returns an error if the file cannot be read or the content is not a valid PID.
146pub fn read_pid_file(path: &str) -> std::io::Result<u32> {
147    let expanded = expand_tilde(path);
148    let content = std::fs::read_to_string(&expanded)?;
149    content
150        .trim()
151        .parse::<u32>()
152        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
153}
154
155/// Remove the PID file.
156///
157/// # Errors
158///
159/// Returns an error if the file cannot be removed.
160pub fn remove_pid_file(path: &str) -> std::io::Result<()> {
161    let expanded = expand_tilde(path);
162    match std::fs::remove_file(&expanded) {
163        Ok(()) => Ok(()),
164        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
165        Err(e) => Err(e),
166    }
167}
168
169fn expand_tilde(path: &str) -> String {
170    if let Some(rest) = path.strip_prefix("~/")
171        && let Some(home) = std::env::var_os("HOME").or_else(|| std::env::var_os("USERPROFILE"))
172    {
173        return format!("{}/{rest}", home.to_string_lossy());
174    }
175    path.to_owned()
176}
177
178#[cfg(test)]
179mod tests {
180    #![allow(clippy::field_reassign_with_default)]
181
182    use super::*;
183
184    #[test]
185    fn expand_tilde_with_home() {
186        let result = expand_tilde("~/test/file.pid");
187        assert!(!result.starts_with("~/"));
188    }
189
190    #[test]
191    fn expand_tilde_absolute_unchanged() {
192        assert_eq!(expand_tilde("/tmp/zeph.pid"), "/tmp/zeph.pid");
193    }
194
195    #[test]
196    fn pid_file_roundtrip() {
197        let dir = tempfile::tempdir().unwrap();
198        let path = dir.path().join("test.pid");
199        let path_str = path.to_string_lossy().to_string();
200
201        write_pid_file(&path_str).unwrap();
202        let pid = read_pid_file(&path_str).unwrap();
203        assert_eq!(pid, std::process::id());
204        remove_pid_file(&path_str).unwrap();
205        assert!(!path.exists());
206    }
207
208    #[test]
209    fn remove_nonexistent_pid_file_ok() {
210        assert!(remove_pid_file("/tmp/nonexistent_zeph_test.pid").is_ok());
211    }
212
213    #[test]
214    fn read_invalid_pid_file() {
215        let dir = tempfile::tempdir().unwrap();
216        let path = dir.path().join("bad.pid");
217        std::fs::write(&path, "not_a_number").unwrap();
218        assert!(read_pid_file(&path.to_string_lossy()).is_err());
219    }
220
221    #[tokio::test]
222    async fn supervisor_tracks_components() {
223        let config = DaemonConfig::default();
224        let (_tx, rx) = watch::channel(false);
225        let mut supervisor = DaemonSupervisor::new(&config, rx);
226
227        let handle = tokio::spawn(async { Ok::<(), DaemonError>(()) });
228        supervisor.add_component(ComponentHandle::new("test", handle));
229        assert_eq!(supervisor.component_count(), 1);
230    }
231
232    #[tokio::test]
233    async fn supervisor_detects_finished_component() {
234        let config = DaemonConfig::default();
235        let (_tx, rx) = watch::channel(false);
236        let mut supervisor = DaemonSupervisor::new(&config, rx);
237
238        let handle = tokio::spawn(async { Ok::<(), DaemonError>(()) });
239        tokio::time::sleep(Duration::from_millis(10)).await;
240        supervisor.add_component(ComponentHandle::new("finished", handle));
241        supervisor.check_health();
242
243        let statuses = supervisor.component_statuses();
244        assert_eq!(statuses.len(), 1);
245        assert!(matches!(statuses[0].1, ComponentStatus::Failed(_)));
246    }
247
248    #[tokio::test]
249    async fn supervisor_shutdown() {
250        let config = DaemonConfig {
251            health_interval_secs: 1,
252            ..DaemonConfig::default()
253        };
254        let (tx, rx) = watch::channel(false);
255        let mut supervisor = DaemonSupervisor::new(&config, rx);
256
257        let run_handle = tokio::spawn(async move { supervisor.run().await });
258        tokio::time::sleep(Duration::from_millis(50)).await;
259        let _ = tx.send(true);
260        tokio::time::timeout(Duration::from_secs(2), run_handle)
261            .await
262            .expect("supervisor should stop on shutdown")
263            .expect("task should complete");
264    }
265
266    #[test]
267    fn component_status_eq() {
268        assert_eq!(ComponentStatus::Running, ComponentStatus::Running);
269        assert_eq!(ComponentStatus::Stopped, ComponentStatus::Stopped);
270        assert_ne!(ComponentStatus::Running, ComponentStatus::Stopped);
271    }
272}
zeph_core/daemon.rs

zeph_core/
daemon.rs
