From bee15ee7cf5a598b72e39d68be35cde9cff988e4 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sat, 19 Sep 2026 14:45:15 +0200 Subject: [PATCH] feat(fuse,packaging): watch the daemon for stalls and answer the systemd watchdog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A supervisor thread ticks every 15 s and does three things: it WARNs about a worker that has held one job for two minutes, a control request that never returned, or a lane that has not completed anything while its queue is non-empty; it samples resident size and the pending-op count into the log every five minutes; and it answers the systemd watchdog. The watchdog ping is deliberately not a heartbeat from a thread that does nothing else — that would keep the unit "healthy" through exactly the hang it is meant to catch. It goes out only after a real round trip over the control socket, so it means "a client can still talk to this daemon", which is the property that failed on 2026-09-18. `sd_notify` is spoken directly rather than through libsystemd (READY, WATCHDOG, STOPPING), including the abstract-namespace address; off systemd every call is a no-op. The unit gains `NotifyAccess=all` (rather than `Type=notify`, which would fail the unit on the start timeout when no login is stored yet), `WatchdogSec=120`, `Restart=always` — a hang is not a failing exit code — and `MemoryHigh`/`MemoryMax` as a backstop against the 5.2 G growth the hung run showed. --- crates/pdfs-fuse/src/lib.rs | 2 + crates/pdfs-fuse/src/mount.rs | 23 +++ crates/pdfs-fuse/src/supervisor.rs | 229 +++++++++++++++++++++++++++++ crates/pdfs-fuse/src/systemd.rs | 99 +++++++++++++ packaging/proton-drive.service | 18 ++- 5 files changed, 370 insertions(+), 1 deletion(-) create mode 100644 crates/pdfs-fuse/src/supervisor.rs create mode 100644 crates/pdfs-fuse/src/systemd.rs diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index c99ede4..b4a97d1 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -96,8 +96,10 @@ mod revisions; mod sharing; mod shutdown; mod state; +mod supervisor; mod sweep; mod sync; +mod systemd; mod takeout; mod transfers; mod upload; diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 53a125d..dca09a6 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -554,6 +554,26 @@ pub fn mount( } } } + // The supervisor watches the pool and answers the systemd watchdog. Started + // after the listener, because its liveness probe is a round trip over that + // socket. + { + let supervisor_core = core.clone(); + let socket = control_socket.to_path_buf(); + match std::thread::Builder::new() + .name("pdfs-supervisor".into()) + .spawn(move || supervisor::run(supervisor_core, socket)) + { + Ok(handle) => workers.push(handle), + // Not fatal: the mount works without a supervisor, it just goes back + // to being undiagnosable if it wedges. + Err(error) => warn!(error = %error, "could not start the daemon supervisor"), + } + } + // The mount is live and the control socket answers: this is the moment the + // unit is actually usable, which is what `Type=notify` reports. + systemd::ready(); + // Re-establish on-demand mounts only after the primary session is live. // Keep the worker so shutdown can close publication, wait for any in-flight // fetch to observe that closure, and then drain a stable registry. @@ -680,6 +700,9 @@ pub fn mount( /// expected and correct: a worker part-way through an upload finishes it rather /// than abandoning the user's bytes mid-flight. fn stop_workers(core: &Core, control_socket: &Path, workers: Vec>) { + // Say so before the joins: a teardown that takes a while must not be read as + // a missed watchdog ping and turned into a kill. + systemd::stopping(); core.shutdown.stop(); core.wake_drain(); let _ = core.sync_tx.send(sync::SyncMsg::Stop); diff --git a/crates/pdfs-fuse/src/supervisor.rs b/crates/pdfs-fuse/src/supervisor.rs new file mode 100644 index 0000000..c7cfdf3 --- /dev/null +++ b/crates/pdfs-fuse/src/supervisor.rs @@ -0,0 +1,229 @@ +//! The thread that watches the daemon for the thing the daemon cannot report +//! about itself: that it has stopped making progress. +//! +//! Three jobs, all cheap, all on one timer: +//! +//! 1. **Say so in the journal.** A worker that has held one job for minutes, a +//! queue that is not draining, a control request that never returned — each +//! gets a WARN naming it. The 2026-09-18 hang left nothing in the journal at +//! all; this is what turns the next one into a one-line diagnosis. +//! 2. **Ping the systemd watchdog**, but only after a real round trip over the +//! control socket. Pinging unconditionally from a thread that does nothing +//! else would keep the unit "healthy" through exactly the hang it is meant to +//! catch: the probe is what makes the ping mean "a client can still talk to +//! this daemon". +//! 3. **Sample memory.** That hang peaked at 5.2 GiB resident, and nobody knew +//! until it was over. + +use std::io::{BufRead, BufReader, Write as _}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use pdfs_core::control::Request as CtlRequest; +use tracing::{info, warn}; + +use super::Core; +use super::systemd; + +/// How often the supervisor looks. Fast enough that a stall is named while the +/// user is still watching it, slow enough to cost nothing. +const TICK: Duration = Duration::from_secs(15); + +/// How long one job may hold a worker, or one control request may run, before it +/// is called out. Generous: a large upload chunk or a cold folder listing on a +/// slow link is minutes of legitimate work, and a false WARN every few minutes +/// would train everyone to ignore the real one. +const STALL_AFTER: Duration = Duration::from_secs(120); + +/// How often resident size goes into the log. +const RSS_EVERY: Duration = Duration::from_secs(300); + +/// How long the liveness probe waits for the daemon to answer its own control +/// socket. Well under the watchdog period, so a probe that times out still +/// leaves time for the manager to hear nothing and act. +const PROBE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Run until shutdown. Spawned once per mount, from [`super::mount::mount`]. +pub(crate) fn run(core: Core, control_socket: PathBuf) { + let watchdog = systemd::watchdog_period(); + if let Some(period) = watchdog { + info!(period_secs = period.as_secs(), "systemd watchdog enabled"); + } + + let mut last_rss = Instant::now(); + let mut last_ping = Instant::now(); + // Per-lane completion counts from the previous tick, to tell a queue that is + // being worked through from one that is not moving at all. + let mut last_completed = (0_u64, 0_u64); + + while core.shutdown.sleep(TICK) { + let report = core.diagnostics(); + let stalled = report_stalls(&report, last_completed); + last_completed = (report.meta_completed, report.transfer_completed); + + if last_rss.elapsed() >= RSS_EVERY { + info!( + rss_bytes = report.rss_bytes, + pending_ops = report.pending_ops, + meta_queued = report.meta_queued, + transfer_queued = report.transfer_queued, + stalled, + "daemon resource sample" + ); + last_rss = Instant::now(); + } + + // A stalled lane is deliberately *not* a reason to withhold the ping: + // the mount may still be answering everything else, and a restart would + // throw away in-flight uploads. The probe decides that; a stall only + // gets said out loud. + if let Some(period) = watchdog + && last_ping.elapsed() >= period + { + if probe(&control_socket) { + systemd::watchdog_ping(); + } else { + warn!( + "daemon did not answer its own control socket; withholding the watchdog ping" + ); + } + last_ping = Instant::now(); + } + } +} + +/// WARN about anything that has not moved for [`STALL_AFTER`]. Returns whether +/// anything was reported, so the caller can weigh it. +fn report_stalls(report: &pdfs_core::control::Diagnostics, last_completed: (u64, u64)) -> bool { + let threshold = STALL_AFTER.as_secs(); + let mut stalled = false; + + for worker in &report.workers { + if worker.busy && worker.age_secs >= threshold { + stalled = true; + warn!( + worker = %worker.name, + job = %worker.label, + age_secs = worker.age_secs, + "a fuse worker has held the same job for a long time" + ); + } + } + if let Some(oldest) = report.inflight.first() + && oldest.age_secs >= threshold + { + stalled = true; + warn!( + request = %oldest.kind, + age_secs = oldest.age_secs, + in_flight = report.inflight.len(), + "a control request has been running for a long time" + ); + } + // A backed-up lane is normal; a backed-up lane that finished nothing since + // the last tick is not. + if report.meta_queued > 0 && report.meta_completed == last_completed.0 { + stalled = true; + warn!( + queued = report.meta_queued, + "the metadata lane has not completed a job since the last check" + ); + } + if report.transfer_queued > 0 && report.transfer_completed == last_completed.1 { + stalled = true; + warn!( + queued = report.transfer_queued, + "the transfer lane has not completed a job since the last check" + ); + } + stalled +} + +/// One round trip over the control socket: connect, ask for the cheapest +/// request there is, read the answer. +/// +/// This is the liveness the watchdog ping stands for. It covers the accept loop, +/// the handler pool and the reply path — everything a `pdfs` command or the GUI +/// depends on — without touching the network or the database. +fn probe(control_socket: &Path) -> bool { + let Ok(stream) = UnixStream::connect(control_socket) else { + return false; + }; + if stream.set_read_timeout(Some(PROBE_TIMEOUT)).is_err() + || stream.set_write_timeout(Some(PROBE_TIMEOUT)).is_err() + { + return false; + } + let Ok(request) = serde_json::to_string(&CtlRequest::Diagnostics) else { + return false; + }; + let mut stream = stream; + if stream.write_all(format!("{request}\n").as_bytes()).is_err() { + return false; + } + let mut line = String::new(); + BufReader::new(stream).read_line(&mut line).is_ok() && !line.trim().is_empty() +} + +#[cfg(test)] +mod tests { + use super::*; + use pdfs_core::control::{Diagnostics, InflightRequest, WorkerState}; + + fn busy_worker(age_secs: u64) -> WorkerState { + WorkerState { + name: "pdfs-fuse-3".into(), + busy: true, + label: "read".into(), + age_secs, + } + } + + /// A worker doing normal work must not be reported: a WARN every tick is a + /// WARN nobody reads. + #[test] + fn ordinary_work_is_not_a_stall() { + let report = Diagnostics { + workers: vec![busy_worker(5)], + ..Default::default() + }; + assert!(!report_stalls(&report, (0, 0))); + } + + /// A worker on the same job past the threshold is the shape of the hang. + #[test] + fn a_long_held_job_is_a_stall() { + let report = Diagnostics { + workers: vec![busy_worker(STALL_AFTER.as_secs() + 1)], + ..Default::default() + }; + assert!(report_stalls(&report, (0, 0))); + } + + /// So is a control request that never returned. + #[test] + fn a_long_running_request_is_a_stall() { + let report = Diagnostics { + inflight: vec![InflightRequest { + kind: "OpenFile".into(), + age_secs: STALL_AFTER.as_secs() + 1, + }], + ..Default::default() + }; + assert!(report_stalls(&report, (0, 0))); + } + + /// A queue that is draining is busy, not stalled — that is what comparing + /// the completion counts is for. + #[test] + fn a_draining_queue_is_not_a_stall() { + let report = Diagnostics { + transfer_queued: 12, + transfer_completed: 40, + ..Default::default() + }; + assert!(!report_stalls(&report, (0, 30))); + assert!(report_stalls(&report, (0, 40))); + } +} diff --git a/crates/pdfs-fuse/src/systemd.rs b/crates/pdfs-fuse/src/systemd.rs new file mode 100644 index 0000000..741df5c --- /dev/null +++ b/crates/pdfs-fuse/src/systemd.rs @@ -0,0 +1,99 @@ +//! The `sd_notify` half of the service contract, without linking libsystemd. +//! +//! The protocol is one datagram of `KEY=VALUE` lines to the socket in +//! `$NOTIFY_SOCKET`. That is little enough to write directly, and writing it +//! directly keeps the daemon buildable on a machine without systemd headers — +//! everything here degrades to "no socket, do nothing" off systemd. + +use std::os::unix::net::UnixDatagram; +use std::time::Duration; + +use tracing::debug; + +/// Send one notification. `false` when there is no manager to send it to, or the +/// send failed — a caller has nothing useful to do about either. +pub(crate) fn notify(message: &str) -> bool { + let Ok(socket_path) = std::env::var("NOTIFY_SOCKET") else { + return false; + }; + if socket_path.is_empty() { + return false; + } + let Ok(socket) = UnixDatagram::unbound() else { + return false; + }; + if let Some(name) = socket_path.strip_prefix('@') { + // Abstract namespace: the address is a NUL byte followed by the name, + // which `std`'s path-based API cannot express. + send_abstract(&socket, name, message) + } else { + socket.send_to(message.as_bytes(), &socket_path).is_ok() + } +} + +/// `sendto` on an abstract-namespace address, which needs a hand-built +/// `sockaddr_un` because the first byte of the name is NUL. +fn send_abstract(socket: &UnixDatagram, name: &str, message: &str) -> bool { + use std::os::fd::AsRawFd; + + let mut addr: libc::sockaddr_un = unsafe { std::mem::zeroed() }; + addr.sun_family = libc::AF_UNIX as libc::sa_family_t; + let bytes = name.as_bytes(); + // One byte for the leading NUL, and the path must still be addressable. + if bytes.len() + 1 > addr.sun_path.len() { + return false; + } + for (slot, byte) in addr.sun_path.iter_mut().skip(1).zip(bytes) { + *slot = *byte as libc::c_char; + } + let len = std::mem::size_of::() + 1 + bytes.len(); + let sent = unsafe { + libc::sendto( + socket.as_raw_fd(), + message.as_ptr().cast(), + message.len(), + 0, + std::ptr::from_ref(&addr).cast(), + len as libc::socklen_t, + ) + }; + sent >= 0 +} + +/// Tell the manager the daemon is up and serving. Called once the mount is live, +/// so `systemctl --user start` returns when the mount actually works rather than +/// when the process exists. +pub(crate) fn ready() { + if notify("READY=1\n") { + debug!("notified systemd: ready"); + } +} + +/// Tell the manager the daemon is still healthy. +pub(crate) fn watchdog_ping() { + let _ = notify("WATCHDOG=1\n"); +} + +/// Tell the manager the daemon is stopping, so a slow teardown is not mistaken +/// for a hang by the watchdog it is racing. +pub(crate) fn stopping() { + let _ = notify("STOPPING=1\n"); +} + +/// How often the manager expects a ping, from `$WATCHDOG_USEC`, or `None` when +/// the watchdog is off or belongs to another process. +/// +/// Reports *half* the configured interval, the interval systemd documents for +/// pinging: a ping period equal to the deadline would restart the daemon on the +/// first late scheduling. +pub(crate) fn watchdog_period() -> Option { + // `WATCHDOG_PID` is set when the manager wants a specific process to answer; + // in that case a ping from anyone else is not the one it is waiting for. + if let Ok(pid) = std::env::var("WATCHDOG_PID") + && pid.parse::().ok() != Some(std::process::id()) + { + return None; + } + let usec: u64 = std::env::var("WATCHDOG_USEC").ok()?.parse().ok()?; + (usec > 0).then(|| Duration::from_micros(usec / 2)) +} diff --git a/packaging/proton-drive.service b/packaging/proton-drive.service index 6677747..94d5517 100644 --- a/packaging/proton-drive.service +++ b/packaging/proton-drive.service @@ -6,12 +6,28 @@ After=graphical-session.target PartOf=graphical-session.target [Service] +# Deliberately not Type=notify: the daemon waits for a stored login before it +# mounts anything, and a machine that boots before the user logs in would then +# fail the unit on the start timeout. NotifyAccess=all gets the daemon a +# NOTIFY_SOCKET all the same, which is all the watchdog below needs. Type=simple +NotifyAccess=all # The daemon waits for a stored login, mounts ~/ProtonDrive, and keeps it # mounted. SIGTERM (from `systemctl --user stop`) triggers a clean lazy unmount. ExecStart=/usr/bin/pdfs daemon -Restart=on-failure +# Always, not on-failure: a daemon that hangs is not a failure by exit code, and +# on 2026-09-18 one had to be restarted by hand after wedging overnight. +Restart=always RestartSec=3 +# The supervisor thread pings this, but only after a round trip over the control +# socket — so a daemon that stops answering its own clients is restarted instead +# of sitting there looking alive. It pings at half this interval. +WatchdogSec=120 +# Backstop against unbounded growth: the hung run peaked at 5.2G resident. +# MemoryHigh throttles and reclaims first; MemoryMax is the hard stop, above any +# legitimate working set, so it only fires on a leak. +MemoryHigh=2G +MemoryMax=6G # Give the lazy unmount time to finish before SIGKILL. TimeoutStopSec=15