rust_web_server/systemd/
mod.rs

1//! systemd integration (18.20.0, #67), without libsystemd.
2//!
3//! - **Notifications** (`Type=notify`): [`ready`] once rws serves, [`reloading`]
4//!   / [`ready`] around a reload, [`stopping`] at shutdown, [`status`] lines,
5//!   and `WATCHDOG=1` every half `WatchdogSec` while the runtime runs.
6//!   Datagrams to `$NOTIFY_SOCKET` (a path, or `@name` in the abstract
7//!   namespace); nothing happens without it.
8//! - **Socket activation**: listeners systemd passes (`LISTEN_FDS`,
9//!   `LISTEN_PID`, `LISTEN_FDNAMES`) are used instead of binding: [`tcp`] and
10//!   [`udp`] return a copy of the inherited socket for an address, matched by
11//!   port (and kind) — systemd's `ListenStream=443` is `[::]:443` while rws's
12//!   `ip` may say `0.0.0.0`. systemd keeps the sockets across restarts, so
13//!   nothing is refused while rws starts again.
14
15use std::net::SocketAddr;
16use std::os::fd::{FromRawFd, RawFd};
17use std::sync::{Mutex, OnceLock};
18
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20enum Kind {
21    Tcp,
22    Udp,
23}
24
25#[derive(Debug)]
26struct Inherited {
27    fd: RawFd,
28    kind: Kind,
29    addr: SocketAddr,
30    name: Option<String>,
31}
32
33/// The sockets systemd passed, read (and the variables removed, so child
34/// processes don't think they're activated) once.
35fn inherited() -> &'static Mutex<Vec<Inherited>> {
36    static SOCKETS: OnceLock<Mutex<Vec<Inherited>>> = OnceLock::new();
37    SOCKETS.get_or_init(|| {
38        let found = read_inherited();
39        if !found.is_empty() {
40            let list: Vec<String> = found.iter().map(|s| format!("{} {:?}{}", s.addr, s.kind, s.name.as_deref().map(|n| format!(" ({n})")).unwrap_or_default())).collect();
41            println!("systemd passed {} socket(s): {}", found.len(), list.join(", "));
42        }
43        Mutex::new(found)
44    })
45}
46
47fn read_inherited() -> Vec<Inherited> {
48    let pid_matches = std::env::var("LISTEN_PID").ok().and_then(|p| p.parse::<u32>().ok()) == Some(std::process::id());
49    let count = std::env::var("LISTEN_FDS").ok().and_then(|n| n.parse::<i32>().ok()).unwrap_or(0);
50    let names: Vec<String> = std::env::var("LISTEN_FDNAMES").map(|n| n.split(':').map(str::to_string).collect()).unwrap_or_default();
51    for var in ["LISTEN_PID", "LISTEN_FDS", "LISTEN_FDNAMES"] {
52        // SAFETY of remove_var: read once at startup, before threads that
53        // read these exist.
54        unsafe { std::env::remove_var(var) };
55    }
56    if !pid_matches || count <= 0 {
57        return Vec::new();
58    }
59    const FIRST: RawFd = 3; // SD_LISTEN_FDS_START
60    let mut out = Vec::new();
61    for (i, fd) in (FIRST..FIRST + count.min(64)).enumerate() {
62        // SAFETY: fds 3.. are ours by the activation protocol.
63        unsafe { libc::fcntl(fd, libc::F_SETFD, libc::FD_CLOEXEC) };
64        let mut kind: libc::c_int = 0;
65        let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
66        // SAFETY: a valid out pointer and length.
67        if unsafe { libc::getsockopt(fd, libc::SOL_SOCKET, libc::SO_TYPE, &mut kind as *mut _ as *mut libc::c_void, &mut len) } != 0 {
68            continue;
69        }
70        let kind = match kind {
71            libc::SOCK_STREAM => Kind::Tcp,
72            libc::SOCK_DGRAM => Kind::Udp,
73            _ => continue,
74        };
75        // The address without taking ownership: borrow through a socket we
76        // forget again.
77        let addr = match kind {
78            Kind::Tcp => {
79                // SAFETY: fd is a stream socket; ownership is given back by into_raw_fd.
80                let socket = unsafe { std::net::TcpListener::from_raw_fd(fd) };
81                let addr = socket.local_addr();
82                std::mem::forget(socket);
83                addr
84            }
85            Kind::Udp => {
86                // SAFETY: as above, for a datagram socket.
87                let socket = unsafe { std::net::UdpSocket::from_raw_fd(fd) };
88                let addr = socket.local_addr();
89                std::mem::forget(socket);
90                addr
91            }
92        };
93        let Ok(addr) = addr else { continue };
94        out.push(Inherited { fd, kind, addr, name: names.get(i).cloned().filter(|n| !n.is_empty() && n != "unknown") });
95    }
96    out
97}
98
99/// The inherited socket of `kind` for `addr` (or named `name`): the same
100/// port, the exact address preferred.
101fn find(kind: Kind, addr: SocketAddr, name: Option<&str>) -> Option<RawFd> {
102    let sockets = inherited().lock().unwrap_or_else(|e| e.into_inner());
103    let candidates = sockets.iter().filter(|s| s.kind == kind);
104    let by_name = name.and_then(|n| candidates.clone().find(|s| s.name.as_deref() == Some(n)));
105    by_name
106        .or_else(|| candidates.clone().find(|s| s.addr == addr))
107        .or_else(|| candidates.clone().find(|s| s.addr.port() == addr.port()))
108        .map(|s| s.fd)
109}
110
111/// A copy of the TCP listener systemd passed for `addr` (or named `name`),
112/// if any. Each call is a new descriptor of the same socket, so several
113/// runtimes can accept on it.
114pub fn tcp(addr: SocketAddr, name: Option<&str>) -> Option<std::net::TcpListener> {
115    let fd = find(Kind::Tcp, addr, name)?;
116    // SAFETY: dup of a listening stream socket we hold.
117    let copy = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 3) };
118    (copy >= 0).then(|| unsafe { std::net::TcpListener::from_raw_fd(copy) })
119}
120
121/// A copy of the UDP socket systemd passed for `addr` (or named `name`).
122pub fn udp(addr: SocketAddr, name: Option<&str>) -> Option<std::net::UdpSocket> {
123    let fd = find(Kind::Udp, addr, name)?;
124    // SAFETY: dup of a datagram socket we hold.
125    let copy = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 3) };
126    (copy >= 0).then(|| unsafe { std::net::UdpSocket::from_raw_fd(copy) })
127}
128
129/// Whether systemd passed any socket.
130pub fn activated() -> bool {
131    !inherited().lock().unwrap_or_else(|e| e.into_inner()).is_empty()
132}
133
134/// Send `message` to `$NOTIFY_SOCKET`; `false` without one (or on error).
135pub fn notify(message: &str) -> bool {
136    let Ok(target) = std::env::var("NOTIFY_SOCKET") else { return false };
137    let Ok(socket) = std::os::unix::net::UnixDatagram::unbound() else { return false };
138    let sent = match target.strip_prefix('@') {
139        #[cfg(target_os = "linux")]
140        Some(name) => {
141            use std::os::linux::net::SocketAddrExt;
142            std::os::unix::net::SocketAddr::from_abstract_name(name.as_bytes()).and_then(|addr| socket.send_to_addr(message.as_bytes(), &addr))
143        }
144        #[cfg(not(target_os = "linux"))]
145        Some(_) => return false,
146        None => socket.send_to(message.as_bytes(), &target),
147    };
148    sent.is_ok()
149}
150
151/// rws serves: `READY=1` (once), and the watchdog starts.
152pub fn ready(status_line: &str) {
153    if notify(&format!("READY=1\nSTATUS={status_line}")) {
154        start_watchdog();
155    }
156}
157
158/// A reload starts (`RELOADING=1` with the monotonic time systemd wants).
159pub fn reloading() {
160    let mut now = libc::timespec { tv_sec: 0, tv_nsec: 0 };
161    // SAFETY: a valid out pointer.
162    unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut now) };
163    let usec = now.tv_sec as u64 * 1_000_000 + now.tv_nsec as u64 / 1_000;
164    notify(&format!("RELOADING=1\nMONOTONIC_USEC={usec}"));
165}
166
167/// A reload is done.
168pub fn reloaded() {
169    notify("READY=1");
170}
171
172/// Shutting down.
173pub fn stopping() {
174    notify("STOPPING=1");
175}
176
177/// A status line for `systemctl status`.
178pub fn status(line: &str) {
179    notify(&format!("STATUS={line}"));
180}
181
182/// `WATCHDOG=1` every half `WATCHDOG_USEC`, from the async runtime when
183/// there is one (a wedged runtime then stops the pings and systemd restarts
184/// rws), else from a thread.
185fn start_watchdog() {
186    static STARTED: OnceLock<()> = OnceLock::new();
187    let Some(usec) = std::env::var("WATCHDOG_USEC").ok().and_then(|u| u.parse::<u64>().ok()).filter(|u| *u > 0) else { return };
188    if std::env::var("WATCHDOG_PID").ok().and_then(|p| p.parse::<u32>().ok()).is_some_and(|p| p != std::process::id()) {
189        return;
190    }
191    if STARTED.set(()).is_err() {
192        return;
193    }
194    let every = std::time::Duration::from_micros(usec / 2);
195    #[cfg(feature = "http2")]
196    if let Ok(handle) = tokio::runtime::Handle::try_current() {
197        handle.spawn(async move {
198            loop {
199                notify("WATCHDOG=1");
200                tokio::time::sleep(every).await;
201            }
202        });
203        return;
204    }
205    let _ = std::thread::Builder::new().name("rws-watchdog".into()).spawn(move || loop {
206        notify("WATCHDOG=1");
207        std::thread::sleep(every);
208    });
209}
210
211#[cfg(test)]
212mod tests;
rust_web_server/systemd/mod.rs

rust_web_server/systemd/
mod.rs
