decdn_node/
systemd_watchdog.rs

1//! Runtime-liveness heartbeat for the systemd service watchdog.
2//!
3//! When the unit sets `WatchdogSec=`, systemd passes `WATCHDOG_USEC` and
4//! `NOTIFY_SOCKET` to the daemon and restarts it if no `WATCHDOG=1` arrives
5//! within that period. The heartbeat is a task on the tokio runtime, so it
6//! proves that a runtime worker is free to run work, not only that the process
7//! exists. A task that holds every worker without awaiting stops the heartbeat,
8//! and systemd then restarts the node instead of reporting it `active` while it
9//! serves nothing.
10//!
11//! The node has no setting for this. The unit's `WatchdogSec=` turns it on and
12//! sets the period. Without `WATCHDOG_USEC` the heartbeat does not start.
13
14use std::ffi::OsString;
15use std::time::Duration;
16
17/// Where to send the heartbeat, and how often.
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct Heartbeat {
20    /// The `NOTIFY_SOCKET` address: a filesystem path, or an abstract socket
21    /// name that starts with `@`.
22    pub socket: OsString,
23    /// Half the watchdog period, as `sd_watchdog_enabled(3)` recommends.
24    pub interval: Duration,
25}
26
27impl Heartbeat {
28    /// The heartbeat systemd asks for through the environment, or `None` when
29    /// it asks for none.
30    ///
31    /// `WATCHDOG_PID`, when set, names the process the watchdog is for. A
32    /// process with another PID does not send the heartbeat.
33    #[must_use]
34    pub fn from_env_vars(
35        notify_socket: Option<OsString>,
36        watchdog_usec: Option<&str>,
37        watchdog_pid: Option<&str>,
38        own_pid: u32,
39    ) -> Option<Self> {
40        let socket = notify_socket.filter(|s| !s.is_empty())?;
41        let usec: u64 = watchdog_usec?.trim().parse().ok()?;
42        if usec == 0 {
43            return None;
44        }
45        if let Some(pid) = watchdog_pid
46            && pid.trim().parse::<u32>().ok()? != own_pid
47        {
48            return None;
49        }
50        let interval = Duration::from_micros(usec / 2).max(Duration::from_millis(1));
51        Some(Self { socket, interval })
52    }
53
54    /// The heartbeat systemd asks this process for.
55    #[must_use]
56    pub fn from_env() -> Option<Self> {
57        Self::from_env_vars(
58            std::env::var_os("NOTIFY_SOCKET"),
59            std::env::var("WATCHDOG_USEC").ok().as_deref(),
60            std::env::var("WATCHDOG_PID").ok().as_deref(),
61            std::process::id(),
62        )
63    }
64}
65
66/// A datagram socket bound to the `NOTIFY_SOCKET` address, built once and
67/// reused for every heartbeat.
68#[cfg(unix)]
69#[derive(Debug)]
70pub struct Notifier {
71    sock: std::os::unix::net::UnixDatagram,
72    addr: std::os::unix::net::SocketAddr,
73}
74
75#[cfg(unix)]
76impl Notifier {
77    /// A notifier for the `NOTIFY_SOCKET` address `socket`: a filesystem path,
78    /// or an abstract socket name that starts with `@`.
79    ///
80    /// # Errors
81    ///
82    /// Fails when the socket cannot be created or the address is invalid.
83    pub fn new(socket: &std::ffi::OsStr) -> std::io::Result<Self> {
84        use std::os::unix::ffi::OsStrExt;
85        use std::os::unix::net::{SocketAddr, UnixDatagram};
86
87        let addr = match socket.as_bytes().strip_prefix(b"@") {
88            Some(name) => abstract_addr(name)?,
89            None => SocketAddr::from_pathname(socket)?,
90        };
91        let sock = UnixDatagram::unbound()?;
92        // A full receive queue must not block a runtime worker.
93        sock.set_nonblocking(true)?;
94        Ok(Self { sock, addr })
95    }
96
97    /// Send one `WATCHDOG=1` datagram.
98    ///
99    /// # Errors
100    ///
101    /// Fails when the datagram cannot be sent.
102    pub fn notify(&self) -> std::io::Result<()> {
103        self.sock.send_to_addr(b"WATCHDOG=1", &self.addr).map(drop)
104    }
105}
106
107#[cfg(target_os = "linux")]
108fn abstract_addr(name: &[u8]) -> std::io::Result<std::os::unix::net::SocketAddr> {
109    use std::os::linux::net::SocketAddrExt;
110    std::os::unix::net::SocketAddr::from_abstract_name(name)
111}
112
113#[cfg(all(unix, not(target_os = "linux")))]
114fn abstract_addr(_name: &[u8]) -> std::io::Result<std::os::unix::net::SocketAddr> {
115    Err(std::io::Error::new(
116        std::io::ErrorKind::Unsupported,
117        "abstract NOTIFY_SOCKET names exist only on Linux",
118    ))
119}
120
121/// Start the heartbeat on the current tokio runtime when systemd asks for one.
122///
123/// Call it from inside the runtime before any other work, so the heartbeat also
124/// covers bring-up and shutdown drain. The task runs until the runtime stops.
125pub fn spawn() {
126    #[cfg(unix)]
127    if let Some(heartbeat) = Heartbeat::from_env() {
128        tokio::spawn(run(heartbeat));
129    }
130}
131
132#[cfg(unix)]
133async fn run(heartbeat: Heartbeat) {
134    let mut ticker = tokio::time::interval(heartbeat.interval);
135    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
136    let mut notifier = None;
137    let mut warned = false;
138    loop {
139        ticker.tick().await;
140        // A notifier that failed to build is built again on the next tick.
141        let sent = match &notifier {
142            Some(n) => Notifier::notify(n),
143            None => Notifier::new(&heartbeat.socket).and_then(|n| {
144                let sent = n.notify();
145                notifier = Some(n);
146                sent
147            }),
148        };
149        match sent {
150            Ok(()) => warned = false,
151            Err(error) if !warned => {
152                warned = true;
153                tracing::warn!(%error, "systemd watchdog heartbeat failed");
154            }
155            Err(_) => {}
156        }
157    }
158}
159
160#[cfg(test)]
161mod tests;
decdn_node/systemd_watchdog.rs

decdn_node/
systemd_watchdog.rs
