feat(fuse,core,cli): report what the daemon is doing, for the next hang

On 2026-09-18 the daemon hung: 43 minutes of CPU over 20 hours, 5.2 GiB
resident, silent from 17:10 until a manual restart the next day, and no
panic. Nothing in the journal said what it had been waiting on, and
nothing in the daemon could be asked.

- The worker pool publishes per-thread state: whether the thread is busy,
  how long it has been in that state, and the label of the job in hand.
  Every FUSE handler that hands work to the pool now names its job
  (`read`, `lookup`, `rename`), so a stuck worker is named rather than
  merely counted. Queue depths and per-lane completion counts are
  mirrored outside the queue lock.
- `Request::Diagnostics` / `pdfs diagnostics` reports those, plus
  resident size, the control handler count against its limit, the
  in-flight control requests with their ages, and the pending-op count.
- Built to answer while the daemon is wedged: atomics and `try_lock`
  only, and the one field that needs the database reads it on a throwaway
  thread with a 500 ms deadline.

crates/pdfs-cli/src/main.rs

@@ -281,6 +281,10 @@ enum Command {
     },
     /// Show the daemon's in-flight transfers (active uploads/downloads).
     Transfers,
+    /// Show what the daemon is doing right now: worker threads, queue depths,
+    /// in-flight control requests and memory. For a daemon that has stopped
+    /// answering rather than one that is merely busy.
+    Diagnostics,
     /// List what is in the account's trash, with the uids the restore and
     /// delete-forever commands take.
     Trash,
@@ -674,6 +678,7 @@ fn main() -> Result<()> {
         Command::Mkdir { parent, name } => cmd_mkdir(parent, name),
         Command::Upload { sources, parent } => cmd_upload(sources, parent),
         Command::Transfers => cmd_transfers(),
+        Command::Diagnostics => cmd_diagnostics(),
         Command::Trash => cmd_trash(),
         Command::Restore { uids } => cmd_restore(uids),
         Command::DeleteForever { uids } => cmd_delete_forever(uids),
@@ -2153,6 +2158,67 @@ fn cmd_rm(path: PathBuf) -> Result<()> {
     Ok(())
 }
 
+/// Seconds as something a human reads at a glance: the ages here run from
+/// "still starting" to "wedged since yesterday".
+fn human_secs(secs: u64) -> String {
+    match secs {
+        s if s < 60 => format!("{s}s"),
+        s if s < 3600 => format!("{}m{}s", s / 60, s % 60),
+        s if s < 86400 => format!("{}h{}m", s / 3600, (s % 3600) / 60),
+        s => format!("{}d{}h", s / 86400, (s % 86400) / 3600),
+    }
+}
+
+fn cmd_diagnostics() -> Result<()> {
+    let response = control_request(CtlRequest::Diagnostics)?;
+    if emit_json(&response)? {
+        return Ok(());
+    }
+    match response {
+        CtlResponse::Diagnostics { diagnostics: d } => {
+            println!(
+                "uptime {}  rss {}  pending ops {}",
+                human_secs(d.uptime_secs),
+                human_bytes(d.rss_bytes),
+                d.pending_ops,
+            );
+            println!(
+                "control handlers {}/{}",
+                d.control_handlers, d.control_handler_limit,
+            );
+            println!(
+                "queued: meta {} (done {}), transfer {} (done {})",
+                d.meta_queued, d.meta_completed, d.transfer_queued, d.transfer_completed,
+            );
+            println!("workers:");
+            for worker in &d.workers {
+                let what = if worker.busy {
+                    worker.label.as_str()
+                } else {
+                    "idle"
+                };
+                println!(
+                    "  {:<20} {:<12} {}",
+                    worker.name,
+                    what,
+                    human_secs(worker.age_secs)
+                );
+            }
+            if d.inflight.is_empty() {
+                println!("in-flight requests: none");
+            } else {
+                println!("in-flight requests (oldest first):");
+                for request in &d.inflight {
+                    println!("  {:<20} {}", request.kind, human_secs(request.age_secs));
+                }
+            }
+        }
+        CtlResponse::Error { message, kind } => bail!("{}", cli_error(kind, &message)),
+        other => bail!("unexpected response: {other:?}"),
+    }
+    Ok(())
+}
+
 fn cmd_trash() -> Result<()> {
     match control_request(CtlRequest::ListTrash)? {
         CtlResponse::Entries { entries } if entries.is_empty() => println!("(trash is empty)"),

crates/pdfs-core/src/control.rs

@@ -299,6 +299,13 @@ pub enum Request {
     /// [`Response::Transfers`]. Cheap to poll: the daemon keeps the registry in
     /// memory, so a front-end can render a live progress widget.
     GetQueueStatus,
+    /// Report what the daemon is doing *right now*, for debugging a daemon that
+    /// stopped answering rather than one that is merely busy. Replies
+    /// [`Response::Diagnostics`].
+    ///
+    /// Answered from lock-free state on purpose: it has to come back while the
+    /// daemon is wedged, which is the only moment anyone asks for it.
+    Diagnostics,
     /// List what is in the account's trash. Replies with [`Response::Entries`];
     /// a trashed node has no path inside the mount, so each entry carries only
     /// its `uid` — the handle for [`Request::Restore`] and
@@ -1400,6 +1407,55 @@ pub fn pending_summary(uploads: u64, changes: u64) -> Option<String> {
     }
 }
 
+/// One worker thread in a [`Diagnostics`] report.
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct WorkerState {
+    /// Thread name (`pdfs-fuse-meta-0`, `pdfs-fuse-3`).
+    pub name: String,
+    pub busy: bool,
+    /// The job in hand (`read`, `lookup`), empty when idle.
+    pub label: String,
+    /// How long it has been in that state. A busy worker with a large age is
+    /// the shape of a hang.
+    pub age_secs: u64,
+}
+
+/// One control request the daemon is still serving, in a [`Diagnostics`] report.
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct InflightRequest {
+    /// The request's variant name (`OpenFile`, `PhotosTimeline`).
+    pub kind: String,
+    pub age_secs: u64,
+}
+
+/// What the daemon is doing right now (reply to [`Request::Diagnostics`]).
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
+pub struct Diagnostics {
+    /// How long this daemon process has been running.
+    pub uptime_secs: u64,
+    /// Resident set size, or `0` where it could not be read. The 2026-09-18 hang
+    /// peaked at 5.2 GiB, so this is the number that says whether the daemon is
+    /// growing again.
+    pub rss_bytes: u64,
+    pub workers: Vec<WorkerState>,
+    /// Jobs waiting for a worker, per lane.
+    pub meta_queued: usize,
+    pub transfer_queued: usize,
+    /// Jobs finished per lane since start. Queue depth that does not move while
+    /// these do not either is a stalled lane, not a loaded one.
+    pub meta_completed: u64,
+    pub transfer_completed: u64,
+    /// Control requests still being served, oldest first.
+    pub inflight: Vec<InflightRequest>,
+    /// Control handler threads in use, and the ceiling. At the ceiling the
+    /// daemon refuses new connections, which looks exactly like a dead daemon
+    /// from the CLI.
+    pub control_handlers: usize,
+    pub control_handler_limit: usize,
+    /// Queued mutations not yet drained.
+    pub pending_ops: u64,
+}
+
 /// The daemon's reply to a [`Request`].
 #[derive(Serialize, Deserialize, Debug)]
 pub enum Response {
@@ -1508,6 +1564,8 @@ pub enum Response {
         available: bool,
         items: Vec<AlbumInfo>,
     },
+    /// What the daemon is doing right now (reply to [`Request::Diagnostics`]).
+    Diagnostics { diagnostics: Diagnostics },
     /// How the Takeout import is doing (reply to [`Request::ImportStatus`]).
     /// `running` is false once it has finished, when `summary` is the final
     /// report; `summary` is `None` when no import has run this session.

crates/pdfs-fuse/src/control.rs

@@ -67,6 +67,16 @@ impl Drop for ControlHandlerPermit {
     }
 }
 
+/// Handler threads in use, for the diagnostics. At the limit the daemon refuses
+/// connections, which a front-end cannot tell apart from a dead daemon.
+pub(crate) fn active_handlers() -> usize {
+    ACTIVE_CONTROL_HANDLERS.load(Ordering::Acquire)
+}
+
+pub(crate) fn handler_limit() -> usize {
+    MAX_CONTROL_HANDLERS
+}
+
 fn read_request_line(reader: &mut impl BufRead) -> std::io::Result<Option<String>> {
     let mut bytes = Vec::new();
     loop {
@@ -178,6 +188,9 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             return;
         }
     };
+    // Registered before the request is even parsed, so a handler that wedges
+    // inside the daemon shows up by name in `pdfs diagnostics`.
+    let _inflight = super::diagnostics::InflightGuard::new(super::diagnostics::request_kind(&line));
     let response = match serde_json::from_str::<CtlRequest>(line.trim()) {
         Ok(CtlRequest::Status) => {
             let pins = core.cache.list_pins();
@@ -657,6 +670,9 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
                 Err(e) => CtlResponse::error(e),
             }
         }
+        Ok(CtlRequest::Diagnostics) => CtlResponse::Diagnostics {
+            diagnostics: core.diagnostics(),
+        },
         Ok(CtlRequest::ListTrash) => match core.list_trash() {
             Ok(entries) => CtlResponse::Entries { entries },
             Err(e) => CtlResponse::error(e),

crates/pdfs-fuse/src/diagnostics.rs

@@ -0,0 +1,212 @@
+//! What the daemon is doing right now, for the case where it has stopped doing
+//! anything at all.
+//!
+//! On 2026-09-18 the daemon hung: it consumed 43 minutes of CPU over 20 hours,
+//! peaked at 5.2 GiB resident, went silent at 17:10, and only came back after a
+//! manual `systemctl --user restart`. There was no panic, so nothing in the
+//! journal said what it had been waiting on — and nothing in the daemon could be
+//! asked, either. This module is the answer to "what was it stuck on?".
+//!
+//! Everything here is built to work *while* the daemon is wedged, which rules
+//! out the obvious implementations: no lock that a stuck job might hold, no
+//! database query on the calling thread, no waiting on the worker pool. A
+//! diagnostic that blocks on the hang it is diagnosing reports nothing at
+//! exactly the moment it matters.
+
+use std::sync::LazyLock;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Instant;
+
+use parking_lot::Mutex;
+use pdfs_core::control::{Diagnostics, InflightRequest, WorkerState};
+
+use super::Core;
+
+/// When this process started. Read at the first touch, which
+/// [`super::mount::mount`] makes happen during startup so the uptime is the
+/// daemon's, not the first diagnostics request's.
+static STARTED: LazyLock<Instant> = LazyLock::new(Instant::now);
+
+/// Control requests currently being served, keyed by a monotonic ticket.
+///
+/// A `Vec` rather than a map because it holds at most `MAX_CONTROL_HANDLERS`
+/// entries, and the only operations are push, remove-by-ticket and read-all.
+static INFLIGHT: Mutex<Vec<(u64, String, Instant)>> = Mutex::new(Vec::new());
+static NEXT_TICKET: AtomicU64 = AtomicU64::new(0);
+
+/// Start the daemon's uptime clock. Called once, from the mount path.
+pub(crate) fn start_clock() {
+    LazyLock::force(&STARTED);
+}
+
+/// Registers one in-flight control request for as long as it is held.
+pub(crate) struct InflightGuard(u64);
+
+impl InflightGuard {
+    pub(crate) fn new(kind: String) -> Self {
+        let ticket = NEXT_TICKET.fetch_add(1, Ordering::Relaxed);
+        INFLIGHT.lock().push((ticket, kind, Instant::now()));
+        Self(ticket)
+    }
+}
+
+impl Drop for InflightGuard {
+    fn drop(&mut self) {
+        INFLIGHT.lock().retain(|(ticket, _, _)| *ticket != self.0);
+    }
+}
+
+/// The variant name of a serialised [`pdfs_core::control::Request`], without
+/// parsing it into one.
+///
+/// serde's external tagging makes this a property of the text: a unit variant is
+/// the bare string `"Status"`, anything else is an object whose single key is
+/// the variant name. Reading the name off the line rather than matching on 80
+/// variants keeps this cheap, and keeps it correct when a variant is added.
+pub(crate) fn request_kind(line: &str) -> String {
+    let line = line.trim();
+    let name = if let Some(rest) = line.strip_prefix('{') {
+        rest.trim_start()
+            .strip_prefix('"')
+            .and_then(|rest| rest.split('"').next())
+    } else {
+        line.strip_prefix('"')
+            .and_then(|rest| rest.split('"').next())
+    };
+    name.unwrap_or("unknown").to_string()
+}
+
+/// Resident set size in bytes, or `0` when `/proc` did not answer.
+///
+/// From `VmRSS` in `/proc/self/status`, which is already in kB — `statm` would
+/// need the page size, and this needs no dependency at all.
+pub(crate) fn rss_bytes() -> u64 {
+    let Ok(status) = std::fs::read_to_string("/proc/self/status") else {
+        return 0;
+    };
+    parse_vm_rss(&status)
+}
+
+/// The `VmRSS` line of a `/proc/<pid>/status`, in bytes.
+fn parse_vm_rss(status: &str) -> u64 {
+    status
+        .lines()
+        .find_map(|line| line.strip_prefix("VmRSS:"))
+        .and_then(|rest| rest.split_whitespace().next())
+        .and_then(|kb| kb.parse::<u64>().ok())
+        .map_or(0, |kb| kb.saturating_mul(1024))
+}
+
+impl Core {
+    /// Assemble a [`Diagnostics`] report.
+    ///
+    /// The pending-op count is the one field that needs the database, so it is
+    /// read on a throwaway thread with a short deadline: a daemon wedged *on*
+    /// the database must still be able to report everything else about itself.
+    pub(crate) fn diagnostics(&self) -> Diagnostics {
+        let pool = self.workers.snapshot();
+        let now = Instant::now();
+        let mut inflight: Vec<InflightRequest> = INFLIGHT
+            .try_lock()
+            .map(|entries| {
+                entries
+                    .iter()
+                    .map(|(_, kind, started)| InflightRequest {
+                        kind: kind.clone(),
+                        age_secs: now.duration_since(*started).as_secs(),
+                    })
+                    .collect()
+            })
+            .unwrap_or_default();
+        inflight.sort_by_key(|request| std::cmp::Reverse(request.age_secs));
+
+        Diagnostics {
+            uptime_secs: STARTED.elapsed().as_secs(),
+            rss_bytes: rss_bytes(),
+            workers: pool
+                .workers
+                .into_iter()
+                .map(|worker| WorkerState {
+                    name: worker.name,
+                    busy: worker.busy,
+                    label: worker.label,
+                    age_secs: worker.age_ms / 1000,
+                })
+                .collect(),
+            meta_queued: pool.meta_queued,
+            transfer_queued: pool.transfer_queued,
+            meta_completed: pool.meta_completed,
+            transfer_completed: pool.transfer_completed,
+            inflight,
+            control_handlers: super::control::active_handlers(),
+            control_handler_limit: super::control::handler_limit(),
+            pending_ops: self.pending_ops_quickly(),
+        }
+    }
+
+    /// Queued mutations, or `0` if the database did not answer within
+    /// [`PENDING_COUNT_DEADLINE`].
+    fn pending_ops_quickly(&self) -> u64 {
+        let db = self.db.clone();
+        let (tx, rx) = std::sync::mpsc::channel();
+        if std::thread::Builder::new()
+            .name("pdfs-diagnostics".into())
+            .spawn(move || {
+                let counts = db.pending_op_counts().unwrap_or_default();
+                let total = counts.uploads.max(0) as u64 + counts.changes.max(0) as u64;
+                let _ = tx.send(total);
+            })
+            .is_err()
+        {
+            return 0;
+        }
+        rx.recv_timeout(PENDING_COUNT_DEADLINE).unwrap_or(0)
+    }
+}
+
+/// How long the report waits for the database before reporting without it.
+const PENDING_COUNT_DEADLINE: std::time::Duration = std::time::Duration::from_millis(500);
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The request name is read off the wire format, so it must match both
+    /// shapes serde produces.
+    #[test]
+    fn the_request_kind_comes_off_both_wire_shapes() {
+        assert_eq!(request_kind("\"Status\""), "Status");
+        assert_eq!(request_kind("{\"Pin\":{\"path\":\"a\"}}"), "Pin");
+        assert_eq!(request_kind(" { \"Restore\" : {} } "), "Restore");
+        assert_eq!(request_kind("garbage"), "unknown");
+    }
+
+    /// A guard registers for its lifetime and nothing longer: a handler that
+    /// returns must not leave a permanent "in flight" entry behind.
+    #[test]
+    fn a_guard_registers_only_while_it_lives() {
+        let before = INFLIGHT.lock().len();
+        {
+            let _guard = InflightGuard::new("OpenFile".into());
+            assert_eq!(INFLIGHT.lock().len(), before + 1);
+        }
+        assert_eq!(INFLIGHT.lock().len(), before);
+    }
+
+    /// Whatever else it reports, the resident size has to be a real reading.
+    #[test]
+    fn rss_is_readable() {
+        assert!(rss_bytes() > 0);
+    }
+
+    /// `VmRSS` is in kB, and a status without one must read as unknown rather
+    /// than as a daemon using no memory.
+    #[test]
+    fn vm_rss_is_kilobytes_and_absence_is_zero() {
+        assert_eq!(
+            parse_vm_rss("VmPeak:\t 100 kB\nVmRSS:\t  2048 kB\n"),
+            2048 * 1024
+        );
+        assert_eq!(parse_vm_rss("VmPeak:\t 100 kB\n"), 0);
+    }
+}

crates/pdfs-fuse/src/filesystem.rs

@@ -49,7 +49,7 @@ impl Filesystem for ProtonFs {
     /// ([`Core::account_quota_cached`] serves a warm one from memory).
     fn statfs(&self, _req: &Request, _ino: INodeNo, reply: ReplyStatfs) {
         let core = self.core.clone();
-        self.core.workers.run(Lane::Meta, move || {
+        self.core.workers.run(Lane::Meta, "statfs", move || {
             let (max_space, used_space) = core.account_quota_cached().unwrap_or((0, 0));
             let bsize = u64::from(REPORTED_BLKSIZE);
             let blocks = (max_space.max(0) as u64).div_ceil(bsize);
@@ -89,7 +89,7 @@ impl Filesystem for ProtonFs {
             return;
         }
         let fs = self.clone();
-        self.core.workers.run(Lane::Meta, move || {
+        self.core.workers.run(Lane::Meta, "lookup", move || {
             fs.serve_lookup(parent, &name, reply, true)
         });
     }
@@ -183,7 +183,7 @@ impl Filesystem for ProtonFs {
         let fs = self.clone();
         self.core
             .workers
-            .run(Lane::Meta, move || fs.serve_opendir(ino, reply));
+            .run(Lane::Meta, "opendir", move || fs.serve_opendir(ino, reply));
     }
 
     fn readdir(
@@ -210,9 +210,9 @@ impl Filesystem for ProtonFs {
             return;
         }
         let fs = self.clone();
-        self.core
+        self.core.workers.run(Lane::Meta, "readdir", move || {
-            .workers
+            fs.serve_readdir(ino, offset, reply)
-            .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply));
+        });
     }
 
     fn releasedir(
@@ -237,9 +237,9 @@ impl Filesystem for ProtonFs {
             return;
         }
         let fs = self.clone();
-        self.core
+        self.core.workers.run(Lane::Transfer, "open", move || {
-            .workers
+            fs.serve_open(ino, flags, reply)
-            .run(Lane::Transfer, move || fs.serve_open(ino, flags, reply));
+        });
     }
 
     fn read(
@@ -329,7 +329,7 @@ impl Filesystem for ProtonFs {
         // even read off the FUSE device. It only reads `state`, so moving it
         // races with nothing. FUSE does not require replies in request order.
         let core = self.core.clone();
-        self.core.workers.run(Lane::Transfer, move || {
+        self.core.workers.run(Lane::Transfer, "read", move || {
             match core.read_range(&uid, mtime, fsize, offset, size as u64, cache_blocks) {
                 Ok(bytes) => reply.data(&bytes),
                 Err(e) => reply.error(e),
@@ -366,9 +366,9 @@ impl Filesystem for ProtonFs {
         // one create at a time, with `ls` on the same mount timing out for as
         // long as the burst lasted. See [`Workers`].
         let fs = self.clone();
-        self.core
+        self.core.workers.run(Lane::Meta, "create", move || {
-            .workers
+            fs.serve_create(parent, &name, reply)
-            .run(Lane::Meta, move || fs.serve_create(parent, &name, reply));
+        });
     }
 
     fn write(
@@ -506,7 +506,7 @@ impl Filesystem for ProtonFs {
                 // the scratch file).
                 let ino = ino.0;
                 let fs = self.clone();
-                self.core.workers.run(Lane::Transfer, move || {
+                self.core.workers.run(Lane::Transfer, "truncate", move || {
                     if let Err(e) = fs.core.queue_truncate(ino, size) {
                         reply.error(e);
                         return;
@@ -620,7 +620,7 @@ impl Filesystem for ProtonFs {
         let fs = self.clone();
         self.core
             .workers
-            .run(Lane::Transfer, move || fs.serve_fsync(fh, reply));
+            .run(Lane::Transfer, "fsync", move || fs.serve_fsync(fh, reply));
     }
 
     fn release(
@@ -680,7 +680,7 @@ impl Filesystem for ProtonFs {
         };
         // `trash_child` is a remote call; see `serve_create`.
         let fs = self.clone();
-        self.core.workers.run(Lane::Meta, move || {
+        self.core.workers.run(Lane::Meta, "unlink", move || {
             fs.serve_unlink(parent, &name_str, reply)
         });
     }
@@ -701,9 +701,9 @@ impl Filesystem for ProtonFs {
         // Enumerates the folder to apply rmdir's emptiness rule, then trashes
         // it: two remote calls. See `serve_create`.
         let fs = self.clone();
-        self.core
+        self.core.workers.run(Lane::Meta, "rmdir", move || {
-            .workers
+            fs.serve_rmdir(parent, &name_str, reply)
-            .run(Lane::Meta, move || fs.serve_rmdir(parent, &name_str, reply));
+        });
     }
 
     fn mkdir(
@@ -731,9 +731,9 @@ impl Filesystem for ProtonFs {
         // remote round trip, and inline it blocks every other request on the
         // mount until it answers.
         let fs = self.clone();
-        self.core
+        self.core.workers.run(Lane::Meta, "mkdir", move || {
-            .workers
+            fs.serve_mkdir(parent, &name, reply)
-            .run(Lane::Meta, move || fs.serve_mkdir(parent, &name, reply));
+        });
     }
 
     fn rename(
@@ -774,7 +774,7 @@ impl Filesystem for ProtonFs {
         // move), and the replaced-destination path adds a trash on top. See
         // `serve_create` for why none of that may run on the dispatch loop.
         let fs = self.clone();
-        self.core.workers.run(Lane::Meta, move || {
+        self.core.workers.run(Lane::Meta, "rename", move || {
             fs.serve_rename(parent, &name, newparent, &newname, flags, reply)
         });
     }
@@ -799,7 +799,7 @@ impl Filesystem for ProtonFs {
         // network per call (B5).
         let fs = self.clone();
         let name = name.to_os_string();
-        self.core.workers.run(Lane::Meta, move || {
+        self.core.workers.run(Lane::Meta, "getxattr", move || {
             fs.serve_getxattr(ino, &name, size, reply)
         });
     }
@@ -1099,7 +1099,7 @@ impl ProtonFs {
         if !off_loop {
             let fs = self.clone();
             let name = name.to_string();
-            self.core.workers.run(Lane::Meta, move || {
+            self.core.workers.run(Lane::Meta, "lookup", move || {
                 fs.serve_lookup(parent, &name, reply, true)
             });
             return;

crates/pdfs-fuse/src/lib.rs

@@ -83,6 +83,7 @@ mod background;
 mod control;
 use control::run_control_socket;
 mod devices;
+mod diagnostics;
 mod drain;
 mod filesystem;
 pub use filesystem::ProtonFs;

crates/pdfs-fuse/src/mount.rs

@@ -341,6 +341,9 @@ pub fn mount(
         username,
         sweep_mode,
     } = options;
+    // Start the uptime clock here rather than at the first `pdfs diagnostics`,
+    // so the age it reports is the daemon's own.
+    super::diagnostics::start_clock();
     let (root, online) = fetch_or_recall_root(&client, &rt, &db)?;
     let scope = root.tree_event_scope_id();
     db.mount_upsert_my_files(&mountpoint.to_string_lossy(), &root.uid.to_string(), None)

crates/pdfs-fuse/src/workers.rs

@@ -12,6 +12,8 @@
 
 use std::collections::VecDeque;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::time::Instant;
 
 use parking_lot::{Condvar, Mutex};
 use tracing::warn;
@@ -53,6 +55,13 @@ const META_WORKERS: usize = 3;
 
 type Job = Box<dyn FnOnce() + Send + 'static>;
 
+/// A queued job and the name its caller gave it, so a job that never finishes
+/// can be named in the diagnostics instead of showing up as "a worker is busy".
+struct Queued {
+    label: &'static str,
+    job: Job,
+}
+
 /// Which lane a job belongs in.
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Lane {
@@ -65,12 +74,63 @@ pub(crate) enum Lane {
 
 #[derive(Default)]
 struct Queues {
-    meta: VecDeque<Job>,
+    meta: VecDeque<Queued>,
-    transfer: VecDeque<Job>,
+    transfer: VecDeque<Queued>,
     /// Set when the pool is dropped, to wake and retire every worker.
     closed: bool,
 }
 
+/// What one worker thread is doing, published for [`Workers::snapshot`].
+///
+/// Every field is an atomic (or a lock the worker only ever holds for the length
+/// of a pointer write) because the whole point of the snapshot is to answer
+/// *while the daemon is wedged*: a diagnostic that waits on the same lock the
+/// stuck job holds reports nothing at exactly the moment it is needed.
+struct Slot {
+    name: String,
+    /// True between picking a job up and finishing it.
+    busy: AtomicBool,
+    /// When the current state (busy or idle) began, as milliseconds since the
+    /// pool started. Monotonic, so an age is always a real duration.
+    since_ms: AtomicU64,
+    /// The label of the job in hand. Behind a mutex because a `&'static str` is
+    /// two words and there is no atomic for that; the worker holds it only to
+    /// swap the pointer, and the snapshot never blocks on it (see
+    /// [`Workers::snapshot`]).
+    label: Mutex<&'static str>,
+}
+
+impl Slot {
+    fn new(name: String, now_ms: u64) -> Self {
+        Self {
+            name,
+            busy: AtomicBool::new(false),
+            since_ms: AtomicU64::new(now_ms),
+            label: Mutex::new(""),
+        }
+    }
+}
+
+/// One worker's state at the moment [`Workers::snapshot`] read it.
+pub(crate) struct WorkerSnapshot {
+    pub(crate) name: String,
+    pub(crate) busy: bool,
+    /// How long the worker has been in that state.
+    pub(crate) age_ms: u64,
+    /// The job in hand, or `""` when idle. `"?"` when the label could not be
+    /// read without blocking.
+    pub(crate) label: String,
+}
+
+/// The pool's state at the moment [`Workers::snapshot`] read it.
+pub(crate) struct PoolSnapshot {
+    pub(crate) workers: Vec<WorkerSnapshot>,
+    pub(crate) meta_queued: usize,
+    pub(crate) transfer_queued: usize,
+    pub(crate) meta_completed: u64,
+    pub(crate) transfer_completed: u64,
+}
+
 /// The queues and the two wait sets over them.
 ///
 /// Two condvars, not one, because the workers are not interchangeable: a
@@ -86,6 +146,22 @@ struct Pool {
     meta_cv: Condvar,
     /// Waited on by the general workers.
     general_cv: Condvar,
+    /// Queue depths, mirrored outside the lock so a snapshot can read them
+    /// without waiting on whatever is holding it.
+    meta_queued: AtomicUsize,
+    transfer_queued: AtomicUsize,
+    /// Jobs finished per lane. A depth that stays high while this stays still is
+    /// a stalled lane rather than a busy one.
+    meta_completed: AtomicU64,
+    transfer_completed: AtomicU64,
+    slots: Vec<Slot>,
+    started: Instant,
+}
+
+impl Pool {
+    fn now_ms(&self) -> u64 {
+        self.started.elapsed().as_millis() as u64
+    }
 }
 
 /// Bounded thread pool behind the network-touching FUSE handlers.
@@ -98,21 +174,35 @@ pub(crate) struct Workers {
 
 impl Workers {
     pub(crate) fn new(n: usize) -> std::io::Result<Self> {
+        // Never leave the pool without a general worker, however small `n` is.
+        let meta_workers = META_WORKERS.min(n.saturating_sub(1));
+        let names: Vec<String> = (0..n)
+            .map(|i| {
+                format!(
+                    "pdfs-fuse-{}{i}",
+                    if i < meta_workers { "meta-" } else { "" }
+                )
+            })
+            .collect();
         let inner = Arc::new(Pool {
             queues: Mutex::new(Queues::default()),
             meta_cv: Condvar::new(),
             general_cv: Condvar::new(),
+            meta_queued: AtomicUsize::new(0),
+            transfer_queued: AtomicUsize::new(0),
+            meta_completed: AtomicU64::new(0),
+            transfer_completed: AtomicU64::new(0),
+            slots: names
+                .iter()
+                .map(|name| Slot::new(name.clone(), 0))
+                .collect(),
+            started: Instant::now(),
         });
-        // Never leave the pool without a general worker, however small `n` is.
+        for (i, name) in names.iter().enumerate() {
-        let meta_workers = META_WORKERS.min(n.saturating_sub(1));
-        for i in 0..n {
             let inner = inner.clone();
             let meta_only = i < meta_workers;
             std::thread::Builder::new()
-                .name(format!(
+                .name(name.clone())
-                    "pdfs-fuse-{}{i}",
-                    if meta_only { "meta-" } else { "" }
-                ))
                 .spawn(move || {
                     let cv = if meta_only {
                         &inner.meta_cv
@@ -121,17 +211,20 @@ impl Workers {
                     };
                     loop {
                         let mut q = inner.queues.lock();
-                        let job = loop {
+                        let picked = loop {
                             // A general worker takes transfers first: metadata
                             // already has threads of its own, so draining the
                             // bulk queue is the useful thing for it to do.
-                            let picked = if meta_only {
+                            let taken = if meta_only {
-                                q.meta.pop_front()
+                                q.meta.pop_front().map(|job| (Lane::Meta, job))
                             } else {
-                                q.transfer.pop_front().or_else(|| q.meta.pop_front())
+                                q.transfer
+                                    .pop_front()
+                                    .map(|job| (Lane::Transfer, job))
+                                    .or_else(|| q.meta.pop_front().map(|job| (Lane::Meta, job)))
                             };
-                            if let Some(job) = picked {
+                            if let Some(taken) = taken {
-                                break Some(job);
+                                break Some(taken);
                             }
                             if q.closed {
                                 break None;
@@ -139,7 +232,15 @@ impl Workers {
                             cv.wait(&mut q);
                         };
                         drop(q);
-                        let Some(job) = job else { break };
+                        let Some((lane, queued)) = picked else { break };
+                        match lane {
+                            Lane::Meta => inner.meta_queued.fetch_sub(1, Ordering::Relaxed),
+                            Lane::Transfer => inner.transfer_queued.fetch_sub(1, Ordering::Relaxed),
+                        };
+                        let slot = &inner.slots[i];
+                        *slot.label.lock() = queued.label;
+                        slot.since_ms.store(inner.now_ms(), Ordering::Relaxed);
+                        slot.busy.store(true, Ordering::Relaxed);
                         // A panicking handler must not cost the pool a worker
                         // for the rest of the run. The dropped `Reply` answers
                         // EIO on its own, so the caller of the failed op is
@@ -164,7 +265,16 @@ impl Workers {
                         // poisoning at all. Anything new reached from a job owes
                         // the same check — an SDK-side `.lock().unwrap()`
                         // silently invalidates this comment.
-                        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job));
+                        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(queued.job));
+                        slot.busy.store(false, Ordering::Relaxed);
+                        slot.since_ms.store(inner.now_ms(), Ordering::Relaxed);
+                        *slot.label.lock() = "";
+                        match lane {
+                            Lane::Meta => inner.meta_completed.fetch_add(1, Ordering::Relaxed),
+                            Lane::Transfer => {
+                                inner.transfer_completed.fetch_add(1, Ordering::Relaxed)
+                            }
+                        };
                     }
                 })?;
         }
@@ -179,7 +289,9 @@ impl Workers {
     /// a pending job holds a `Reply` and a few fields, while the 4 MiB block
     /// buffer is allocated inside the job once it runs, and the SDK's in-flight
     /// semaphore is what bounds how many of those exist at once.
-    pub(crate) fn run(&self, lane: Lane, job: impl FnOnce() + Send + 'static) {
+    /// `label` names the work for the diagnostics (`pdfs diagnostics`) and for
+    /// the stall warnings: it is what a hung daemon reports instead of "busy".
+    pub(crate) fn run(&self, lane: Lane, label: &'static str, job: impl FnOnce() + Send + 'static) {
         let mut q = self.inner.queues.lock();
         if q.closed {
             // Pre-pool behaviour: a shut-down pool degrades to a slow mount
@@ -189,9 +301,19 @@ impl Workers {
             job();
             return;
         }
+        let queued = Queued {
+            label,
+            job: Box::new(job),
+        };
         match lane {
-            Lane::Meta => q.meta.push_back(Box::new(job)),
+            Lane::Meta => {
-            Lane::Transfer => q.transfer.push_back(Box::new(job)),
+                q.meta.push_back(queued);
+                self.inner.meta_queued.fetch_add(1, Ordering::Relaxed);
+            }
+            Lane::Transfer => {
+                q.transfer.push_back(queued);
+                self.inner.transfer_queued.fetch_add(1, Ordering::Relaxed);
+            }
         }
         drop(q);
         match lane {
@@ -209,6 +331,41 @@ impl Workers {
             }
         }
     }
+
+    /// What every worker is doing right now, without taking the queue lock.
+    ///
+    /// A diagnostic is worth having exactly when the daemon is stuck, so this
+    /// reads atomics only, and `try_lock`s each label rather than waiting for
+    /// it: a label is held for one pointer write, so failing to get it means a
+    /// racing worker, not a wedged one, and `"?"` is a better answer than a
+    /// diagnostics request that hangs too.
+    pub(crate) fn snapshot(&self) -> PoolSnapshot {
+        let now = self.inner.now_ms();
+        let workers = self
+            .inner
+            .slots
+            .iter()
+            .map(|slot| {
+                let label = slot
+                    .label
+                    .try_lock()
+                    .map_or_else(|| "?".to_string(), |label| (*label).to_string());
+                WorkerSnapshot {
+                    name: slot.name.clone(),
+                    busy: slot.busy.load(Ordering::Relaxed),
+                    age_ms: now.saturating_sub(slot.since_ms.load(Ordering::Relaxed)),
+                    label,
+                }
+            })
+            .collect();
+        PoolSnapshot {
+            workers,
+            meta_queued: self.inner.meta_queued.load(Ordering::Relaxed),
+            transfer_queued: self.inner.transfer_queued.load(Ordering::Relaxed),
+            meta_completed: self.inner.meta_completed.load(Ordering::Relaxed),
+            transfer_completed: self.inner.transfer_completed.load(Ordering::Relaxed),
+        }
+    }
 }
 
 impl Drop for Workers {
@@ -225,6 +382,35 @@ mod tests {
     use std::sync::mpsc;
     use std::time::Duration;
 
+    /// A diagnostic is only useful if it names the work. A running job must show
+    /// up under its label, and the queue behind it must show up as depth.
+    #[test]
+    fn a_snapshot_names_the_running_job_and_the_queue_behind_it() {
+        let pool = Workers::new(2).unwrap();
+        let (release_tx, release_rx) = mpsc::channel::<()>();
+        let release_rx = Arc::new(Mutex::new(release_rx));
+        let (started_tx, started_rx) = mpsc::channel();
+        {
+            let release_rx = release_rx.clone();
+            pool.run(Lane::Transfer, "read", move || {
+                started_tx.send(()).unwrap();
+                let _ = release_rx.lock().recv();
+            });
+        }
+        started_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("the job starts");
+
+        let snapshot = pool.snapshot();
+        assert!(
+            snapshot.workers.iter().any(|w| w.busy && w.label == "read"),
+            "the running job is named"
+        );
+        assert_eq!(snapshot.transfer_completed, 0);
+
+        let _ = release_tx.send(());
+    }
+
     /// Audit A6. Transfers filling every general worker must not delay a
     /// metadata job: that is what the reserved threads are for.
     #[test]
@@ -240,7 +426,7 @@ mod tests {
         for _ in 0..general {
             let release_rx = release_rx.clone();
             let started_tx = started_tx.clone();
-            pool.run(Lane::Transfer, move || {
+            pool.run(Lane::Transfer, "test", move || {
                 started_tx.send(()).unwrap();
                 let _ = release_rx.lock().recv();
             });
@@ -255,13 +441,13 @@ mod tests {
         // in the way that used to starve metadata.
         for _ in 0..16 {
             let release_rx = release_rx.clone();
-            pool.run(Lane::Transfer, move || {
+            pool.run(Lane::Transfer, "test", move || {
                 let _ = release_rx.lock().recv();
             });
         }
 
         let (meta_tx, meta_rx) = mpsc::channel();
-        pool.run(Lane::Meta, move || meta_tx.send(()).unwrap());
+        pool.run(Lane::Meta, "test", move || meta_tx.send(()).unwrap());
         meta_rx
             .recv_timeout(Duration::from_secs(5))
             .expect("a metadata job runs while every transfer thread is stuck");
@@ -279,7 +465,7 @@ mod tests {
         // must have been served by the general worker.
         let pool = Workers::new(1).unwrap();
         let (tx, rx) = mpsc::channel();
-        pool.run(Lane::Meta, move || tx.send(()).unwrap());
+        pool.run(Lane::Meta, "test", move || tx.send(()).unwrap());
         rx.recv_timeout(Duration::from_secs(5))
             .expect("a general worker takes metadata when no transfer is waiting");
     }
@@ -288,9 +474,9 @@ mod tests {
     #[test]
     fn a_panicking_job_does_not_retire_its_worker() {
         let pool = Workers::new(1).unwrap();
-        pool.run(Lane::Meta, || panic!("boom"));
+        pool.run(Lane::Meta, "test", || panic!("boom"));
         let (tx, rx) = mpsc::channel();
-        pool.run(Lane::Meta, move || tx.send(()).unwrap());
+        pool.run(Lane::Meta, "test", move || tx.send(()).unwrap());
         rx.recv_timeout(Duration::from_secs(5))
             .expect("the pool still serves work after a panic");
     }