Merge pull request #80 from narrrl/fix/stale-mount-watchdog-startup

fix(fuse,cli): recover the mount after a watchdog kill and bound thumbnail memory

crates/pdfs-cli/src/main.rs

@@ -2520,7 +2520,10 @@ fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> {
     // An explicit `--mountpoint` wins; otherwise honor the Settings-page choice
     // (config), falling back to the default location.
     let mountpoint = mountpoint.unwrap_or_else(|| dirs.resolved_mountpoint(&config));
-    std::fs::create_dir_all(&mountpoint)
+    // Hydrating a large account and remounting its folders can outlast the
+    // unit's watchdog before the supervisor sends its first ping (B279).
+    pdfs_fuse::stretch_watchdog_for_startup();
+    pdfs_fuse::prepare_mountpoint(&mountpoint)
         .with_context(|| format!("create mountpoint {}", mountpoint.display()))?;
 
     let username = auth::load().map(|s| s.username).unwrap_or_default();

crates/pdfs-fuse/src/lib.rs

@@ -119,13 +119,14 @@ mod workers;
 use background::{forget_and_notify, run_event_sync, run_local_index, run_photos_event_sync};
 use link::BATCH_CALL_TIMEOUT;
 pub(crate) use mount::is_stale_mount;
+pub use mount::prepare_mountpoint;
 pub use mount::{MountOptions, MountOutcome, mount};
 use mount::{
     SecondaryInsertRejection, SecondaryMount, clear_stale_mount, fuse_connection_id, spawn_session,
 };
 use reads::{BlockFlight, BlockRing, PREFETCH_BUDGET, Prefetch, ReaderSlot, STREAM_BYPASS_MIN};
 use state::{Entry, Intervals, PendingRevision, State, StateGuard, WriteHandle};
-pub use systemd::{idle_sleep, status as notify_status};
+pub use systemd::{idle_sleep, status as notify_status, stretch_watchdog_for_startup};
 use tracing::{debug, error, info, warn};
 use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry};
 use r#virtual::{

crates/pdfs-fuse/src/mount.rs

@@ -196,6 +196,20 @@ pub(super) fn clear_stale_mount(mountpoint: &Path) {
     }
 }
 
+/// Make sure `mountpoint` is a directory a mount can go on, clearing a dead
+/// mount a killed daemon left there first.
+///
+/// A dead mount answers `ENOTCONN`, which `create_dir_all` takes for "absent";
+/// it then tries `mkdir` and fails with `EEXIST`. The daemon retried that error
+/// forever and never got as far as [`check_mountpoint`], which would have
+/// cleared the mount (`docs/BUGS.md` B279).
+pub fn prepare_mountpoint(mountpoint: &Path) -> std::io::Result<()> {
+    if is_stale_mount(mountpoint) {
+        clear_stale_mount(mountpoint);
+    }
+    std::fs::create_dir_all(mountpoint)
+}
+
 /// Refuse to mount My files over a folder that already has files in it: FUSE
 /// would hide them, unsent and out of sight, until the service stops
 /// (`docs/BUGS.md` B277). Runs before the root fetch and before any worker

crates/pdfs-fuse/src/photos.rs

@@ -72,6 +72,105 @@ const UPLOAD_THUMB_QUALITIES: [u8; 4] = [65, 50, 35, 20];
 /// Largest still image read whole to render the thumbnail its upload carries.
 const UPLOAD_IMAGE_MAX_BYTES: u64 = 256 * 1024 * 1024;
 
+/// How much memory every thumbnail decode in the daemon may hold at once:
+/// the encoded file plus its decoded pixels.
+///
+/// Without a limit, each of the 16 drain workers decoded its own full-size
+/// photo. A batch of 100-megapixel JPEGs then held about 9 GiB, which swapped the
+/// machine until the watchdog killed the daemon, on every restart
+/// (`docs/BUGS.md` B280). With this budget, a phone photo still decodes beside
+/// several others, and a 100-megapixel one decodes alone.
+const DECODE_BUDGET_BYTES: u64 = 512 * 1024 * 1024;
+
+/// The process-wide [`DecodeBudget`] every thumbnail decode draws from.
+static DECODE_BUDGET: DecodeBudget = DecodeBudget::new(DECODE_BUDGET_BYTES);
+
+/// A counting gate over bytes: [`acquire`](Self::acquire) waits until the cost
+/// fits beside what is already held.
+struct DecodeBudget {
+    limit: u64,
+    used: std::sync::Mutex<u64>,
+    freed: std::sync::Condvar,
+}
+
+/// A share of a [`DecodeBudget`], given back on drop.
+struct DecodePermit<'a> {
+    budget: &'a DecodeBudget,
+    cost: u64,
+}
+
+impl DecodeBudget {
+    const fn new(limit: u64) -> Self {
+        Self {
+            limit,
+            used: std::sync::Mutex::new(0),
+            freed: std::sync::Condvar::new(),
+        }
+    }
+
+    /// Wait until `cost` bytes fit, and hold them. A cost above the limit is
+    /// cut down to the limit, so the largest image waits for an idle budget
+    /// and then runs alone, instead of waiting forever.
+    fn acquire(&self, cost: u64) -> DecodePermit<'_> {
+        let cost = cost.clamp(1, self.limit);
+        let mut used = self
+            .used
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        while *used + cost > self.limit {
+            used = self
+                .freed
+                .wait(used)
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+        }
+        *used += cost;
+        DecodePermit { budget: self, cost }
+    }
+}
+
+impl Drop for DecodePermit<'_> {
+    fn drop(&mut self) {
+        let mut used = self
+            .budget
+            .used
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        *used -= self.cost;
+        self.budget.freed.notify_all();
+    }
+}
+
+/// What decoding an image of `encoded_len` bytes and `dimensions` costs in
+/// memory: the encoded bytes, the decoded pixels at up to 4 bytes each, and a
+/// second copy of the pixels when [`decode_upright`] rotates them. When the
+/// header gives no dimensions, assume the image is compressed 16 to 1.
+fn decode_cost(encoded_len: u64, dimensions: Option<(u32, u32)>) -> u64 {
+    let pixels = match dimensions {
+        Some((width, height)) => u64::from(width) * u64::from(height) * 4 * 2,
+        None => encoded_len.saturating_mul(16),
+    };
+    encoded_len.saturating_add(pixels)
+}
+
+/// The pixel dimensions in the header of an encoded image, without decoding it.
+fn dimensions_of(bytes: &[u8]) -> Option<(u32, u32)> {
+    image::ImageReader::new(std::io::Cursor::new(bytes))
+        .with_guessed_format()
+        .ok()?
+        .into_dimensions()
+        .ok()
+}
+
+/// [`dimensions_of`] for an image file. Reads only the header.
+fn dimensions_of_file(path: &Path) -> Option<(u32, u32)> {
+    image::ImageReader::open(path)
+        .ok()?
+        .with_guessed_format()
+        .ok()?
+        .into_dimensions()
+        .ok()
+}
+
 /// The aspect ratio (w/h) of an encoded image, read from its header alone — no
 /// pixels are decoded. `None` when the format is unknown or the header is torn.
 fn ratio_of(bytes: &[u8]) -> Option<f64> {
@@ -209,7 +308,14 @@ fn thumbnail_build_may_start(status: &ThumbnailBuildStatus, path: &str) -> CoreR
 /// formats return an undecodable verdict.
 ///
 /// CPU-bound (a 20 MP JPEG is real work), so callers run it on the blocking pool.
+/// Waits for its share of [`DECODE_BUDGET`] first.
 fn scale_thumbnail(bytes: &[u8], name: &str, staging_dir: &Path) -> ScaleAttempt {
+    let _permit = DECODE_BUDGET.acquire(decode_cost(bytes.len() as u64, dimensions_of(bytes)));
+    scale_thumbnail_unbudgeted(bytes, name, staging_dir)
+}
+
+/// [`scale_thumbnail`] for a caller that already holds its [`DecodePermit`].
+fn scale_thumbnail_unbudgeted(bytes: &[u8], name: &str, staging_dir: &Path) -> ScaleAttempt {
     scale_thumbnail_with_exiftool(
         bytes,
         name,
@@ -544,11 +650,20 @@ fn upload_thumbnails_with(
         }
         video_file_thumbnail(path, ffmpeg, FFMPEG_TIMEOUT)
     } else if is_thumbnail_image_name(name) {
-        let fits = std::fs::metadata(path).is_ok_and(|meta| meta.len() <= UPLOAD_IMAGE_MAX_BYTES);
-        let Some(bytes) = fits.then(|| std::fs::read(path).ok()).flatten() else {
+        let Some(len) = std::fs::metadata(path)
+            .ok()
+            .map(|meta| meta.len())
+            .filter(|len| *len <= UPLOAD_IMAGE_MAX_BYTES)
+        else {
             return Vec::new();
         };
-        scale_thumbnail(&bytes, name, staging_dir)
+        // Taken before the read, so the encoded bytes count against the
+        // budget too.
+        let _permit = DECODE_BUDGET.acquire(decode_cost(len, dimensions_of_file(path)));
+        let Ok(bytes) = std::fs::read(path) else {
+            return Vec::new();
+        };
+        scale_thumbnail_unbudgeted(&bytes, name, staging_dir)
     } else {
         return Vec::new();
     };
@@ -2116,6 +2231,66 @@ impl Core {
     }
 }
 
+#[cfg(test)]
+mod decode_budget_tests {
+    use super::{DecodeBudget, decode_cost};
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::time::Duration;
+
+    /// A decode that does not fit beside the ones running waits for them,
+    /// instead of every drain worker decoding at once (`docs/BUGS.md` B280).
+    #[test]
+    fn a_decode_over_the_budget_waits_for_the_running_one() {
+        let budget = Arc::new(DecodeBudget::new(100));
+        let first = budget.acquire(60);
+        let started = Arc::new(AtomicBool::new(false));
+        let waiter = {
+            let (budget, started) = (budget.clone(), started.clone());
+            std::thread::spawn(move || {
+                let _second = budget.acquire(60);
+                started.store(true, Ordering::SeqCst);
+            })
+        };
+        std::thread::sleep(Duration::from_millis(100));
+        assert!(
+            !started.load(Ordering::SeqCst),
+            "second decode ran over budget"
+        );
+        drop(first);
+        waiter.join().expect("waiter");
+        assert!(started.load(Ordering::SeqCst));
+    }
+
+    /// Decodes that fit together run together.
+    #[test]
+    fn small_decodes_share_the_budget() {
+        let budget = DecodeBudget::new(100);
+        let _a = budget.acquire(30);
+        let _b = budget.acquire(30);
+        let _c = budget.acquire(30);
+    }
+
+    /// An image bigger than the whole budget still decodes, alone, rather than
+    /// waiting forever.
+    #[test]
+    fn an_image_over_the_whole_budget_still_decodes() {
+        let budget = DecodeBudget::new(100);
+        drop(budget.acquire(10_000));
+        let _again = budget.acquire(100);
+    }
+
+    /// The cost of the 102-megapixel JPEG that set off B280 fills the real
+    /// budget, so it never decodes beside another one.
+    #[test]
+    fn a_hundred_megapixel_photo_costs_the_whole_budget() {
+        let cost = decode_cost(97_266_141, Some((11_656, 8_742)));
+        assert!(cost >= super::DECODE_BUDGET_BYTES);
+        let phone = decode_cost(4_000_000, Some((4_000, 3_000)));
+        assert!(phone * 4 < super::DECODE_BUDGET_BYTES);
+    }
+}
+
 #[cfg(test)]
 mod thumb_tests {
     use super::{

crates/pdfs-fuse/src/supervisor.rs

@@ -57,7 +57,10 @@ pub(crate) fn run(core: Core, control_socket: PathBuf) {
     }
 
     let mut last_rss = Instant::now();
-    let mut last_ping = Instant::now();
+    // `None` until the first probe, which runs on the first tick: startup has
+    // already used up much of the watchdog's time (`docs/BUGS.md` B279).
+    let mut last_ping: Option<Instant> = None;
+    let mut restored = false;
     // 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);
@@ -84,16 +87,22 @@ pub(crate) fn run(core: Core, control_socket: PathBuf) {
         // 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
+            && ping_due(last_ping, period)
         {
             if probe(&control_socket) {
+                if !restored {
+                    // Startup ran on the longer timeout the daemon asked for.
+                    // The socket answers now, so the unit's own one applies.
+                    systemd::restore_watchdog();
+                    restored = true;
+                }
                 systemd::watchdog_ping();
             } else {
                 warn!(
                     "daemon did not answer its own control socket; withholding the watchdog ping"
                 );
             }
-            last_ping = Instant::now();
+            last_ping = Some(Instant::now());
         }
     }
 }
@@ -165,6 +174,11 @@ fn report_stalls(report: &pdfs_core::control::Diagnostics, last_completed: (u64,
     stalled
 }
 
+/// Whether the watchdog wants its next ping. The first one is due at once.
+fn ping_due(last_ping: Option<Instant>, period: Duration) -> bool {
+    last_ping.is_none_or(|at| at.elapsed() >= period)
+}
+
 /// One round trip over the control socket: connect, ask for the cheapest
 /// request there is, read the answer.
 ///
@@ -205,6 +219,18 @@ mod tests {
         }
     }
 
+    /// The first ping goes out on the first tick, not a full period after the
+    /// supervisor starts: startup has already spent most of the watchdog's time
+    /// (`docs/BUGS.md` B279).
+    #[test]
+    fn first_watchdog_ping_is_due_at_once() {
+        let period = Duration::from_secs(60);
+        assert!(ping_due(None, period));
+        assert!(!ping_due(Some(Instant::now()), period));
+        let long_ago = Instant::now().checked_sub(period).expect("uptime");
+        assert!(ping_due(Some(long_ago), period));
+    }
+
     /// A worker doing normal work must not be reported: a WARN every tick is a
     /// WARN nobody reads.
     #[test]

crates/pdfs-fuse/src/systemd.rs

@@ -102,6 +102,37 @@ pub fn idle_sleep(duration: Duration) {
     }
 }
 
+/// How long a mount may take to start before the watchdog counts it as hung.
+///
+/// Startup has no control socket to probe yet, and on a large account it is
+/// long: hydrating 48 000 nodes took 70 s on 2026-11-12, and the secondary
+/// mounts followed. Under the 2-minute `WatchdogSec` that run was aborted just
+/// before its first ping, and left six dead mounts behind (`docs/BUGS.md` B279).
+const STARTUP_WATCHDOG: Duration = Duration::from_secs(600);
+
+/// Give the mount about to start [`STARTUP_WATCHDOG`] before the watchdog fires.
+///
+/// Sends `WATCHDOG_USEC=`, which replaces the unit's timeout and restarts its
+/// timer. The supervisor puts the configured timeout back with
+/// [`restore_watchdog`] once it can prove the daemon answers. A startup that
+/// wedges for good is still aborted, only later.
+pub fn stretch_watchdog_for_startup() {
+    if watchdog_period().is_some() {
+        notify(&format!("WATCHDOG_USEC={}\n", STARTUP_WATCHDOG.as_micros()));
+    }
+}
+
+/// Put back the timeout the unit configured, undoing
+/// [`stretch_watchdog_for_startup`].
+pub(crate) fn restore_watchdog() {
+    if watchdog_period().is_none() {
+        return;
+    }
+    if let Ok(usec) = std::env::var("WATCHDOG_USEC") {
+        notify(&format!("WATCHDOG_USEC={usec}\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() {

docs/BUGS.md

@@ -109,6 +109,8 @@ a live account since.
 
 | Entry | Title |
 |---|---|
+| [B280](#b280--uploading-a-batch-of-large-photos-swaps-the-machine-until-the-daemon-is-killed) | Uploading a batch of large photos swaps the machine until the daemon is killed |
+| [B279](#b279--after-a-watchdog-kill-the-mount-never-comes-back) | After a watchdog kill the mount never comes back |
 | [B278](#b278--sync-list-and-devices-list-show-the-last-sync-as-epoch-seconds) | `sync list` and `devices list` show the last sync as epoch seconds |
 | [B277](#b277--my-files-mounted-over-a-folder-with-files-in-it-and-hid-them) | My files mounted over a folder with files in it, and hid them |
 | [B274](#b274--opening-a-file-for-writing-fails-with-eio-while-its-queued-revision-lands) | Opening a file for writing fails with EIO while its queued revision lands |
@@ -381,6 +383,76 @@ a live account since.
 
 Newest first.
 
+### B280 — Uploading a batch of large photos swaps the machine until the daemon is killed
+
+**Status:** Fixed, not verified live
+**Found:** 2026-11-12, after B279. 868 new pictures (15.6 GiB) were copied into
+`~/ProtonDrive/Pictures`. Each restart reached 1.7 GiB resident and 9.8 GiB of swap within 40 s.
+The 16 `pdfs-drain-*` threads were all busy, cgroup memory pressure (`full avg10`) was 77 %, and
+the system's 15 GiB of swap was full. The watchdog killed the daemon, and the next start did the
+same. The largest picture was a 97 MB JPEG of 11656×8742 pixels.
+**Where:** `upload_thumbnails_with` and `scale_thumbnail` in `crates/pdfs-fuse/src/photos.rs`
+
+**Repro:** Queue more large photos than there are drain workers, for example by copying a folder
+of 100-megapixel JPEGs into the mount, and start the daemon.
+
+**Cause.** Every upload of an image builds its thumbnail by decoding the whole file. Nothing
+limited how many of these decodes ran at once, so all 16 drain workers did it together. One
+102-megapixel JPEG holds about 300 MB of pixels, and twice that when it has to be turned upright.
+Sixteen of them do not fit in memory, and `MemoryHigh=2G` throttled the daemon into swap until it
+stopped answering.
+
+**Fix.** Each thumbnail decode first takes a share of a process-wide 512 MiB budget. The share is
+the encoded size plus 8 bytes per pixel, read from the image header. A 12-megapixel phone photo
+still decodes beside four others. A 100-megapixel photo waits until no other decode runs, and
+then runs alone. The upload path takes its share before it reads the file.
+
+**Test:** `decode_budget_tests` in `crates/pdfs-fuse/src/photos.rs`. On 2026-11-12 the same queue
+on the fixed build peaked at 1.3 GiB resident and 90 MB of swap, with memory pressure near 0 %.
+
+---
+
+### B279 — After a watchdog kill the mount never comes back
+
+**Status:** Fixed, not verified live
+**Found:** 2026-11-12, copying a folder of pictures into `~/ProtonDrive/Pictures`. The daemon
+stopped answering, the watchdog aborted it after 4 h 39 min (2.5 GiB memory peak, 11.8 GiB swap
+peak), and every restart then logged
+`create mountpoint /home/narl/ProtonDrive: File exists (os error 17)`. After the dead mounts were
+cleared by hand, the next start mounted all six folders and was aborted again, 120 s after it
+began, leaving six dead mounts behind once more.
+**Where:** `mount_once` in `crates/pdfs-cli/src/main.rs`; `prepare_mountpoint` in
+`crates/pdfs-fuse/src/mount.rs`; `crates/pdfs-fuse/src/supervisor.rs`;
+`crates/pdfs-fuse/src/systemd.rs`
+
+**Repro:** Kill the daemon with `SIGABRT` while it is mounted. Each restart fails with `EEXIST`
+until `fusermount3 -uz` clears every mount. Then restart on an account where hydrating the
+metadata cache takes about a minute: the watchdog aborts the new daemon before it pings.
+
+**Cause.** Two faults, one after the other.
+
+1. A dead FUSE mount answers `ENOTCONN`. `create_dir_all` in `mount_once` read that as "absent",
+   tried `mkdir`, and failed with `EEXIST`. The daemon retried that error forever and never
+   reached `check_mountpoint`, which clears a dead mount.
+2. systemd arms the watchdog when the process starts, but the supervisor sent its first ping one
+   full period (60 s) after it started. The supervisor starts only once the mount is live: 74 s
+   in on this account (70 s of that is hydrating 47 958 nodes). 74 s + 60 s is past the 120 s
+   `WatchdogSec`, so the run was aborted just before its first ping.
+
+**Fix.** `mount_once` calls `prepare_mountpoint`, which clears a dead mount before it creates the
+directory. Before startup, the daemon sends `WATCHDOG_USEC=` to allow 10 minutes. When the
+supervisor first gets an answer from the control socket, it sends the unit's own timeout back.
+The supervisor also probes and pings on its first tick, not one period later. A startup that
+hangs for good is still aborted, only later.
+
+The daemon stopped answering during the copy because of
+[B280](#b280--uploading-a-batch-of-large-photos-swaps-the-machine-until-the-daemon-is-killed).
+
+**Test:** `first_watchdog_ping_is_due_at_once` in `crates/pdfs-fuse/src/supervisor.rs`. The
+stale-mount path and the watchdog handshake need a live systemd unit to check.
+
+---
+
 ### B278 — `sync list` and `devices list` show the last sync as epoch seconds
 
 **Status:** Fixed, not verified live