rightkit_daemon/
ticker.rs

1//! Periodic work and a heartbeat record for a daemon host (issue #376).
2//!
3//! [`spawn_ticker`] runs a closure every interval on its own thread until a [`StopToken`] is
4//! triggered. [`spawn_heartbeat`] is a ticker that writes a small JSON record (`pid`, `updated_ms`
5//! and the caller's `state`) atomically to a path, so a supervisor or a client can read it. Both
6//! end on the first tick after the stop and do no work after it. A write failure is reported to
7//! the caller's `on_error` and the heartbeat keeps going.
8
9use crate::record::write_atomic;
10use crate::stop::StopToken;
11use serde_json::{json, Value};
12use std::io;
13use std::path::PathBuf;
14use std::thread::JoinHandle;
15use std::time::{Duration, SystemTime, UNIX_EPOCH};
16
17/// Run `tick` every `interval` until `stop` is triggered. The first tick comes one interval after
18/// the spawn. The thread is named `rk-ticker`.
19pub fn spawn_ticker(
20    interval: Duration,
21    stop: StopToken,
22    mut tick: impl FnMut() + Send + 'static,
23) -> io::Result<JoinHandle<()>> {
24    std::thread::Builder::new()
25        .name("rk-ticker".into())
26        .spawn(move || loop {
27            if stop.wait(interval).is_some() {
28                break;
29            }
30            tick();
31        })
32}
33
34/// The heartbeat record: `pid`, `updated_ms` (milliseconds since the Unix epoch) and `state`.
35pub fn heartbeat_record(state: Value) -> Value {
36    let updated_ms = SystemTime::now()
37        .duration_since(UNIX_EPOCH)
38        .map(|d| d.as_millis() as u64)
39        .unwrap_or(0);
40    json!({"pid": std::process::id(), "updated_ms": updated_ms, "state": state})
41}
42
43/// Write a heartbeat record to `path` now and then every `interval`, until `stop` is triggered.
44/// `state` is called for each write. Writes go through [`write_atomic`], so a reader never sees a
45/// partial record. A failed write is passed to `on_error`; the next interval tries again.
46pub fn spawn_heartbeat(
47    path: PathBuf,
48    interval: Duration,
49    stop: StopToken,
50    state: impl Fn() -> Value + Send + 'static,
51    mut on_error: impl FnMut(&io::Error) + Send + 'static,
52) -> io::Result<JoinHandle<()>> {
53    let write = move |path: &PathBuf, state: &dyn Fn() -> Value| {
54        let body = serde_json::to_vec(&heartbeat_record(state()))
55            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
56        write_atomic(path, &body)
57    };
58    std::thread::Builder::new()
59        .name("rk-heartbeat".into())
60        .spawn(move || loop {
61            if let Err(e) = write(&path, &state) {
62                on_error(&e);
63            }
64            if stop.wait(interval).is_some() {
65                break;
66            }
67        })
68}
69
70#[cfg(test)]
71mod tests {
72    use super::*;
73    use crate::stop::{StopCause, StopSource};
74    use std::sync::atomic::{AtomicU32, Ordering};
75    use std::sync::Arc;
76
77    #[test]
78    fn a_ticker_ticks_until_stopped_and_then_no_more() {
79        let source = StopSource::new();
80        let count = Arc::new(AtomicU32::new(0));
81        let c = count.clone();
82        let handle = spawn_ticker(Duration::from_millis(5), source.token(), move || {
83            c.fetch_add(1, Ordering::SeqCst);
84        })
85        .unwrap();
86        std::thread::sleep(Duration::from_millis(60));
87        source.trigger(StopCause::Requested);
88        handle.join().unwrap();
89        let at_stop = count.load(Ordering::SeqCst);
90        assert!(at_stop >= 1, "the ticker ran");
91        std::thread::sleep(Duration::from_millis(20));
92        assert_eq!(
93            count.load(Ordering::SeqCst),
94            at_stop,
95            "no tick after the stop"
96        );
97    }
98
99    #[test]
100    fn a_heartbeat_record_has_pid_time_and_state() {
101        let record = heartbeat_record(json!({"phase": "ready"}));
102        assert_eq!(record["pid"], std::process::id());
103        assert_eq!(record["state"]["phase"], "ready");
104        assert!(record["updated_ms"].as_u64().unwrap_or(0) > 0);
105    }
106}
rightkit_daemon/ticker.rs

rightkit_daemon/
ticker.rs
