fix(fuse): keep inode stable across rename

relocate() called forget(), dropping the node's by_uid mapping, so
the re-enumeration assigned a fresh inode. The kernel had already
carried the renamed dentry onto the old one and holds it for the
entry TTL, so lookup/getattr/opendir resolved to an inode entries
no longer held and returned ENOENT — while readdir walked children
and reported the new inode. A renamed directory listed in its
parent but could not be entered until the TTL expired.

Rewrite the node in place and invalidate both parents' listings
instead. Also stops the moved node's DB row being deleted and
recreated on every move, which the B1 data-loss fix worked around

crates/pdfs-core/src/cache.rs

@@ -179,6 +179,12 @@ pub struct ContentCache {
     /// **never** emptied on open: these are the only copy of content the user
     /// authored, and the whole point is that they outlive the daemon.
     staging_dir: PathBuf,
+    /// Subdirectory holding scratch files rescued from an unclean shutdown —
+    /// writes an application had `fsync`ed but not yet closed. Populated only at
+    /// open, drained by the daemon into `staging_dir` once it can address the
+    /// nodes they belong to. Like `staging_dir` and unlike `scratch_dir`, it is
+    /// never emptied blindly.
+    recovery_dir: PathBuf,
     /// JSON pin registry path.
     pins_path: PathBuf,
     /// Soft cap on total blob bytes. Exceeded only transiently: a `store`
@@ -225,9 +231,14 @@ impl ContentCache {
         std::fs::create_dir_all(&thumb_dir)?;
         let block_dir = content_dir.join("blocks");
         std::fs::create_dir_all(&block_dir)?;
-        // Scratch holds disk-backed write buffers; a previous run's leftovers are
+        // Scratch holds disk-backed write buffers. A previous run's leftovers are
-        // worthless, so start clean.
+        // worthless *except* where the application asked for durability: a handle
+        // that was `fsync`ed but never closed left a sidecar behind, and POSIX
+        // says those bytes are on stable storage. Rescue those, discard the rest.
         let scratch_dir = content_dir.join("scratch");
+        let recovery_dir = content_dir.join("recovery");
+        std::fs::create_dir_all(&recovery_dir)?;
+        Self::rescue_scratch(&scratch_dir, &recovery_dir);
         let _ = std::fs::remove_dir_all(&scratch_dir);
         std::fs::create_dir_all(&scratch_dir)?;
         // Staging, by contrast, is deliberately preserved across runs.
@@ -242,6 +253,7 @@ impl ContentCache {
             block_dir,
             scratch_dir,
             staging_dir,
+            recovery_dir,
             pins_path,
             max_bytes: AtomicU64::new(max_bytes),
             blob_bytes: AtomicU64::new(0),
@@ -433,6 +445,36 @@ impl ContentCache {
         Ok(())
     }
 
+    /// Adopt an existing on-disk file as `uid`'s cached content, without reading
+    /// it into memory.
+    ///
+    /// [`Self::store`] takes the bytes as a slice, which means a caller holding a
+    /// file has to load all of it first — fine for a thumbnail, an OOM for a
+    /// pinned video. This takes the path instead and hardlinks it into place,
+    /// falling back to a copy when `src` lives on another filesystem.
+    ///
+    /// The link makes the cache blob a second name for the same inode, so `src`
+    /// must not be rewritten in place afterwards — unlinking it (the staging
+    /// case, which is what this exists for) is fine and leaves the blob intact.
+    pub fn store_file(&self, uid: &NodeUid, mtime: i64, size: u64, src: &Path) -> Result<()> {
+        let blob = self.blob_path(uid);
+        let tmp = blob.with_extension("tmp");
+        // A stale tmp from an interrupted run would fail the link with EEXIST.
+        let _ = std::fs::remove_file(&tmp);
+        if std::fs::hard_link(src, &tmp).is_err() {
+            std::fs::copy(src, &tmp)?;
+        }
+        let bytes = std::fs::metadata(&tmp)?.len();
+        std::fs::rename(&tmp, &blob)?;
+        let meta = serde_json::to_vec(&Meta { mtime, size })?;
+        std::fs::write(self.meta_path(uid), meta)?;
+        self.db
+            .cache_touch(&Self::key(uid), KIND_BLOB, bytes, now_millis())?;
+        self.blob_bytes.fetch_add(bytes, Ordering::Relaxed);
+        self.enforce_budget();
+        Ok(())
+    }
+
     /// Cache-index key for block `idx` of `uid` — the block file's name, which
     /// is also its `cache_entries.cache_key`.
     fn block_key(&self, uid: &NodeUid, idx: u64) -> String {
@@ -626,6 +668,127 @@ impl ContentCache {
         Ok((file, path))
     }
 
+    /// The sidecar path for a scratch file, holding the [`StagedWrite`] that
+    /// describes it. Written by `fsync` and removed when the write is released;
+    /// its presence at open is what marks a scratch file as durable rather than
+    /// as a crashed run's rubbish.
+    ///
+    /// Safe as `with_extension` only because scratch names are dot-free (see
+    /// [`Self::create_scratch`]), so this appends rather than replaces.
+    pub fn scratch_sidecar(scratch: &Path) -> PathBuf {
+        scratch.with_extension("json")
+    }
+
+    /// Record that the write buffered in `scratch` is durable, so a crash before
+    /// `close(2)` does not lose it.
+    ///
+    /// Called from `fsync(2)`, whose contract is that the bytes survive a crash.
+    /// The scratch file itself does survive — it is a real file — but nothing
+    /// else knows it holds anything worth keeping, and open() clears the scratch
+    /// directory. The sidecar is that knowledge: uid, logical length, and which
+    /// ranges are authored rather than holes, which is exactly what recovery
+    /// needs to hand the blob to the drain.
+    ///
+    /// Written whole to a temp name and renamed, so a crash mid-write leaves the
+    /// previous sidecar rather than a truncated one. The caller is expected to
+    /// have synced `scratch` first: a sidecar promising bytes that never reached
+    /// the disk is worse than no sidecar.
+    pub fn mark_scratch_durable(&self, scratch: &Path, meta: &StagedWrite) -> Result<()> {
+        let side = Self::scratch_sidecar(scratch);
+        let tmp = side.with_extension("json.tmp");
+        std::fs::write(&tmp, serde_json::to_vec(meta)?)?;
+        std::fs::rename(&tmp, &side)?;
+        // The rename is what publishes the sidecar; fsync the directory so the
+        // rename itself survives, not just the bytes it points at.
+        if let Some(dir) = side.parent()
+            && let Ok(d) = std::fs::File::open(dir)
+        {
+            let _ = d.sync_all();
+        }
+        Ok(())
+    }
+
+    /// Drop a scratch file's durability sidecar, once the write it described has
+    /// been staged (or turned out not to be dirty at all). Best effort: a
+    /// leftover sidecar whose blob is gone is ignored by recovery anyway.
+    pub fn clear_scratch_durable(&self, scratch: &Path) {
+        let _ = std::fs::remove_file(Self::scratch_sidecar(scratch));
+    }
+
+    /// Move every scratch file that carries a readable sidecar into `recovery`,
+    /// before the caller empties the scratch directory.
+    ///
+    /// Blob first, then sidecar: a blob in recovery without its sidecar is
+    /// dropped on the next open (indistinguishable from rubbish), whereas a
+    /// sidecar whose blob never arrived would describe nothing. Neither order is
+    /// lossless if we crash mid-rescue, but this one fails toward "unrecovered
+    /// bytes on disk" rather than "op queued against missing content".
+    fn rescue_scratch(scratch_dir: &Path, recovery_dir: &Path) {
+        let Ok(entries) = std::fs::read_dir(scratch_dir) else {
+            return;
+        };
+        for entry in entries.flatten() {
+            let blob = entry.path();
+            if blob.extension().is_some_and(|e| e == "json") {
+                continue;
+            }
+            let side = Self::scratch_sidecar(&blob);
+            // Parsed, not merely present: an unreadable sidecar cannot drive a
+            // recovery, so the blob is rubbish like any other leftover.
+            if std::fs::read(&side)
+                .ok()
+                .and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok())
+                .is_none()
+            {
+                continue;
+            }
+            let Some(name) = blob.file_name() else {
+                continue;
+            };
+            if std::fs::rename(&blob, recovery_dir.join(name)).is_err() {
+                continue;
+            }
+            let Some(side_name) = side.file_name() else {
+                continue;
+            };
+            let _ = std::fs::rename(&side, recovery_dir.join(side_name));
+        }
+    }
+
+    /// Writes rescued from an unclean shutdown, as `(blob, meta)` pairs for the
+    /// daemon to hand to the upload queue. A blob whose sidecar did not survive
+    /// the rescue is deleted here — nothing can address it.
+    pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> {
+        let Ok(entries) = std::fs::read_dir(&self.recovery_dir) else {
+            return Vec::new();
+        };
+        let mut out = Vec::new();
+        for entry in entries.flatten() {
+            let blob = entry.path();
+            if blob.extension().is_some_and(|e| e == "json") {
+                continue;
+            }
+            match std::fs::read(Self::scratch_sidecar(&blob))
+                .ok()
+                .and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok())
+            {
+                Some(meta) => out.push((blob, meta)),
+                None => {
+                    let _ = std::fs::remove_file(&blob);
+                }
+            }
+        }
+        out
+    }
+
+    /// Retire a recovered write. `stage_write` has already moved the blob out,
+    /// so this is the sidecar left behind — and the blob too if the caller could
+    /// not use it and said so by leaving it in place.
+    pub fn discard_recovered(&self, blob: &Path) {
+        let _ = std::fs::remove_file(Self::scratch_sidecar(blob));
+        let _ = std::fs::remove_file(blob);
+    }
+
     /// Move a released scratch file into the staging directory with a
     /// [`StagedWrite`] sidecar, and return where it landed.
     ///
@@ -1072,6 +1235,47 @@ mod tests {
         );
     }
 
+    /// The drain promotes a staged upload into the cache by path and then
+    /// unlinks the staging name. The cached copy has to survive that, and the
+    /// promote has to charge the LRU index like [`ContentCache::store`] does.
+    #[test]
+    fn store_file_adopts_blob_and_survives_source_unlink() {
+        let (c, d) = cache();
+        let u = uid("a");
+        let data = b"staged contents";
+        let src = d.path().join("staged-blob");
+        std::fs::write(&src, data).unwrap();
+
+        c.store_file(&u, 100, data.len() as u64, &src).unwrap();
+        std::fs::remove_file(&src).unwrap();
+
+        assert!(c.is_cached(&u, 100, data.len() as u64));
+        assert_eq!(
+            c.read_range(&u, 100, data.len() as u64, 7, 8).unwrap(),
+            b"contents"
+        );
+        assert_eq!(c.usage(), data.len() as u64);
+    }
+
+    /// A cross-filesystem staging dir cannot be hardlinked from; the copy
+    /// blob and its metadata must both be replaced wholesale — a stale `.meta`
+    /// left beside new content would validate a read of the wrong bytes. The
+    /// link goes to a temp name and is renamed over the blob for exactly this,
+    /// so re-storing over an existing entry is the case worth pinning.
+    #[test]
+    fn store_file_replaces_an_existing_blob() {
+        let (c, d) = cache();
+        let u = uid("a");
+        c.store(&u, 100, 3, b"old").unwrap();
+
+        let src = d.path().join("new-blob");
+        std::fs::write(&src, b"newer").unwrap();
+        c.store_file(&u, 200, 5, &src).unwrap();
+
+        assert!(!c.is_cached(&u, 100, 3), "stale meta must not validate");
+        assert_eq!(c.read_range(&u, 200, 5, 0, 5).unwrap(), b"newer");
+    }
+
     #[test]
     fn reconcile_rebuilds_index_from_disk() {
         let dir = TempDir::new();
@@ -1210,13 +1414,12 @@ mod tests {
             c.store_block(&uid("a"), 1, 100, i, b"aaaa").unwrap();
             std::thread::sleep(std::time::Duration::from_millis(2));
         }
-        let indexed: u64 = c
+        let indexed: u64 =
-            .db
+            c.db.cache_entries_by_kind("block")
-            .cache_entries_by_kind("block")
+                .unwrap()
-            .unwrap()
+                .iter()
-            .iter()
+                .map(|(_, s)| s)
-            .map(|(_, s)| s)
+                .sum();
-            .sum();
         assert_eq!(
             c.block_bytes.load(Ordering::Relaxed),
             indexed,
@@ -1351,7 +1554,10 @@ mod tests {
         if let Some(got) = c.read_thumbnail(&u, 2, 100) {
             assert_eq!(got, two, "type 2 served type 1's bytes");
         }
-        assert_eq!(store_errors, 0, "a store lost its staging file to the other type");
+        assert_eq!(
+            store_errors, 0,
+            "a store lost its staging file to the other type"
+        );
     }
 
     #[test]
@@ -1577,6 +1783,89 @@ mod tests {
         );
     }
 
+    /// A2: `fsync(2)` promises the bytes survive a crash, but queueing happens
+    /// at `release`. A scratch file marked durable must therefore outlive the
+    /// open that clears the scratch directory — and an unmarked one must not,
+    /// since that is a crashed run's rubbish.
+    #[test]
+    fn fsynced_scratch_survives_reopen_and_unmarked_scratch_does_not() {
+        use std::io::Write as _;
+        let dir = TempDir::new();
+        let db = Arc::new(Db::open_in_memory().unwrap());
+        let content = dir.path().join("content");
+        let pins = dir.path().join("pins.json");
+        let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap();
+
+        let (mut synced, synced_path) = c.create_scratch().unwrap();
+        synced.write_all(b"fsynced bytes").unwrap();
+        synced.sync_all().unwrap();
+        let meta = StagedWrite {
+            uid: uid("a").to_string(),
+            len: 13,
+            base_size: 0,
+            base_mtime: 100,
+            authored: vec![(0, 13)],
+            complete: true,
+            based_on: None,
+        };
+        c.mark_scratch_durable(&synced_path, &meta).unwrap();
+
+        // A second handle that was written but never fsynced: no promise was
+        // made about it, and it is indistinguishable from a torn buffer.
+        let (mut loose, loose_path) = c.create_scratch().unwrap();
+        loose.write_all(b"never fsynced").unwrap();
+
+        // Crash: no release, no staging, just a restart.
+        drop(c);
+        let c2 = ContentCache::open(content, pins, 0, db).unwrap();
+
+        assert!(!synced_path.exists(), "scratch dir is still cleared");
+        assert!(!loose_path.exists());
+
+        let recovered = c2.recovered_writes();
+        assert_eq!(recovered.len(), 1, "only the fsynced write is recoverable");
+        let (blob, got) = &recovered[0];
+        assert_eq!(std::fs::read(blob).unwrap(), b"fsynced bytes");
+        assert_eq!(got.uid, uid("a").to_string());
+        assert_eq!(got.authored, vec![(0, 13)]);
+
+        // Staging it is what a restart does with it; the sidecar goes after.
+        let staged = c2.stage_write(got, blob).unwrap();
+        c2.discard_recovered(blob);
+        assert_eq!(std::fs::read(&staged).unwrap(), b"fsynced bytes");
+        assert!(c2.recovered_writes().is_empty(), "recovery does not repeat");
+    }
+
+    /// Releasing a write hands the bytes to staging, so the sidecar an earlier
+    /// `fsync` left must not offer recovery a second, stale copy of it.
+    #[test]
+    fn cleared_durability_marker_stops_recovery() {
+        use std::io::Write as _;
+        let dir = TempDir::new();
+        let db = Arc::new(Db::open_in_memory().unwrap());
+        let content = dir.path().join("content");
+        let pins = dir.path().join("pins.json");
+        let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap();
+
+        let (mut f, path) = c.create_scratch().unwrap();
+        f.write_all(b"released").unwrap();
+        let meta = StagedWrite {
+            uid: uid("a").to_string(),
+            len: 8,
+            base_size: 0,
+            base_mtime: 100,
+            authored: vec![(0, 8)],
+            complete: true,
+            based_on: None,
+        };
+        c.mark_scratch_durable(&path, &meta).unwrap();
+        c.clear_scratch_durable(&path);
+
+        drop(c);
+        let c2 = ContentCache::open(content, pins, 0, db).unwrap();
+        assert!(c2.recovered_writes().is_empty());
+    }
+
     #[test]
     fn scratch_file_is_writable_and_isolated() {
         use std::io::{Read as _, Seek as _, SeekFrom, Write as _};

crates/pdfs-core/src/config.rs

@@ -1,6 +1,6 @@
 //! Static client identity and per-user filesystem paths.
 
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 
 use directories::ProjectDirs;
 use serde::{Deserialize, Serialize};
@@ -165,18 +165,104 @@ impl AppDirs {
             .unwrap_or_else(|| self.default_mountpoint())
     }
 
-    /// Create state + cache dirs if missing.
+    /// Create state + cache dirs if missing, owner-only.
+    ///
+    /// The mode is set explicitly rather than left to the umask. These
+    /// directories hold decrypted file content, a plaintext index of every node
+    /// name in the Drive, and the control socket — and `create_dir_all` under a
+    /// typical umask makes them `0755`. Until this was fixed, the only thing
+    /// keeping another local user out was `~/.cache` and `~/.local/state`
+    /// happening to be `0700`, which is a convention of the user's system and
+    /// not something this client established (bugs.md B6).
+    ///
+    /// Applied on every start, not just at creation: a directory that already
+    /// exists with a permissive mode — restored from a backup that flattened
+    /// modes, or created by an older build — is tightened here.
     pub fn ensure(&self) -> Result<()> {
-        std::fs::create_dir_all(self.state_dir())?;
+        for dir in [self.state_dir(), self.cache_dir()] {
-        std::fs::create_dir_all(self.cache_dir())?;
+            std::fs::create_dir_all(&dir)?;
-        let _ = std::fs::create_dir_all(self.dirs.config_dir());
+            restrict_dir(&dir);
+        }
+        let config_dir = self.dirs.config_dir().to_path_buf();
+        if std::fs::create_dir_all(&config_dir).is_ok() {
+            restrict_dir(&config_dir);
+        }
         Ok(())
     }
 }
 
+/// Make `dir` owner-only (`0700`). Best effort: a mode we cannot set (an
+/// exotic filesystem, a directory we do not own) is not worth refusing to
+/// start over, and the caller has no better answer than continuing.
+fn restrict_dir(dir: &Path) {
+    use std::os::unix::fs::PermissionsExt;
+    if let Err(e) = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) {
+        tracing::warn!(path = %dir.display(), error = %e, "could not restrict directory to 0700");
+    }
+}
+
+/// Make a just-bound Unix socket owner-only (`0600`).
+///
+/// Connecting to the control socket drives the daemon with its authenticated
+/// session — enumerate, read, upload, trash, share — without presenting any
+/// credential, since the daemon already holds one. It is an authority boundary,
+/// not merely private data, and `UnixListener::bind` applies the umask like any
+/// other file (bugs.md B6).
+///
+/// There is a window between `bind` and this call during which the socket
+/// carries the umask's mode. It is closed in practice by the containing
+/// directory: [`Dirs::ensure`] has already made the state directory `0700`, so
+/// nothing else can reach the socket to exploit the window.
+pub fn restrict_socket(path: &Path) -> Result<()> {
+    use std::os::unix::fs::PermissionsExt;
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::os::unix::fs::PermissionsExt;
+
+    fn mode_of(p: &Path) -> u32 {
+        std::fs::metadata(p).unwrap().permissions().mode() & 0o777
+    }
+
+    /// bugs.md B6. A permissive mode here is not a cosmetic problem: these
+    /// directories hold decrypted content, a plaintext index of every node name,
+    /// and the control socket that commands the daemon's session.
+    #[test]
+    fn restrict_dir_makes_a_directory_owner_only() {
+        let dir = std::env::temp_dir().join(format!("pdfs-cfg-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        // Start deliberately world-readable, as `create_dir_all` under a typical
+        // umask would leave it.
+        std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)).unwrap();
+
+        restrict_dir(&dir);
+
+        assert_eq!(mode_of(&dir), 0o700, "group and other must have no access");
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    /// Connecting to the control socket confers the daemon's authenticated
+    /// session with no credential, so it is an authority boundary rather than
+    /// merely private data.
+    #[test]
+    fn restrict_socket_makes_a_socket_owner_only() {
+        let dir = std::env::temp_dir().join(format!("pdfs-sock-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        let sock = dir.join("control.sock");
+        let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap();
+
+        restrict_socket(&sock).unwrap();
+
+        assert_eq!(mode_of(&sock), 0o600, "no other local user may connect");
+        drop(listener);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 
     #[test]
     fn test_app_config_serialization() {

crates/pdfs-core/src/control.rs

@@ -76,11 +76,18 @@ pub enum Request {
     PhotoThumbs { uids: Vec<String> },
     /// Download a photo's full content into the cache; replies with its path.
     OpenPhoto { uid: String },
-    /// Upload a photo with the given name, media type, and content bytes.
+    /// Upload the photo at `source_path` under the given name and media type.
+    ///
+    /// A path, not the bytes: this protocol is line-delimited JSON, and
+    /// `serde_json` writes a `Vec<u8>` as an array of decimal integers — a ~5-6x
+    /// inflation that both peers then hold in memory at once, which turned a
+    /// large photo or a video into an OOM of the GUI, the daemon, or both. The
+    /// daemon shares a filesystem with every front-end it serves (the socket is
+    /// a Unix socket), so it opens and streams the file itself.
     UploadPhoto {
         name: String,
         media_type: String,
-        bytes: Vec<u8>,
+        source_path: String,
         capture_time: Option<i64>,
     },
     /// Download a Drive file's full content into the cache; replies with the
@@ -106,13 +113,10 @@ pub enum Request {
     /// Create a new folder named `name` under the mountpoint-relative `parent`.
     /// Replies with [`Response::Ok`].
     CreateFolder { parent: String, name: String },
-    /// Upload a file named `name` with content `bytes` into the
+    // There is deliberately no bytes-carrying single-file upload: it was
-    /// mountpoint-relative `parent` folder. Replies with [`Response::Ok`].
+    // `UploadFile { parent, name, bytes }`, and it OOMed on anything large for
-    UploadFile {
+    // the reason described on `UploadPhoto`. `UploadPaths` covers the same case
-        parent: String,
+    // by path, including the one-file batch.
-        name: String,
-        bytes: Vec<u8>,
-    },
     /// Bulk-upload local files and/or directory trees into the mountpoint-relative
     /// `parent` folder. `sources` are absolute paths on the daemon's own
     /// filesystem (the daemon is local): each file is uploaded, each directory is
@@ -1036,10 +1040,11 @@ mod tests {
                 parent: "a".into(),
                 name: "new".into(),
             },
-            Request::UploadFile {
+            Request::UploadPhoto {
-                parent: "a".into(),
+                name: "p.jpg".into(),
-                name: "f.bin".into(),
+                media_type: "image/jpeg".into(),
-                bytes: vec![0, 1, 2, 255],
+                source_path: "/home/u/p.jpg".into(),
+                capture_time: Some(1_700_000_000),
             },
             Request::UploadPaths {
                 parent: "a".into(),

crates/pdfs-core/src/db/cache.rs

@@ -90,7 +90,11 @@ impl Db {
     /// pass over a large cache does not read every row to drop a few, and so the
     /// `(kind, last_accessed)` index can answer it as a range scan with an early
     /// stop.
-    pub fn cache_eviction_candidates(&self, kind: &str, limit: usize) -> Result<Vec<(String, u64)>> {
+    pub fn cache_eviction_candidates(
+        &self,
+        kind: &str,
+        limit: usize,
+    ) -> Result<Vec<(String, u64)>> {
         let conn = self.conn.lock();
         let mut stmt = conn.prepare_cached(
             "SELECT cache_key, size_bytes FROM cache_entries

crates/pdfs-core/src/db/tests.rs

@@ -373,7 +373,8 @@ fn children_if_listed_gated_on_flag() {
 fn a_deleted_child_leaves_a_listed_parent_serving_a_stale_listing() {
     let db = Db::open_in_memory().unwrap();
     db.upsert_node(&folder("root", None, "My Files")).unwrap();
-    db.upsert_node(&folder("dst", Some("root"), "dest")).unwrap();
+    db.upsert_node(&folder("dst", Some("root"), "dest"))
+        .unwrap();
     db.upsert_node(&file("f1", "root", "a.txt", 1)).unwrap();
     db.set_listed(&uid("dst"), true).unwrap();
 
@@ -1302,7 +1303,10 @@ fn next_due_op_respects_backoff() {
     let id = db.enqueue_op(&bulk_op(1, &root)).unwrap().0;
     db.record_op_failure(id, "boom", 5_000).unwrap();
 
-    assert!(db.next_due_op(4_999).unwrap().is_none(), "still backing off");
+    assert!(
+        db.next_due_op(4_999).unwrap().is_none(),
+        "still backing off"
+    );
     assert!(db.next_due_op(5_000).unwrap().is_some(), "due now");
 }
 
@@ -1362,7 +1366,9 @@ fn cache_total_bytes_agrees_with_summing_the_rows() {
     // query. The claim that matters — that the *common* path does not touch the
     // database at all — is pinned by `budget_check_is_free_when_under_budget`,
     // which compares against itself and so is load-independent.
-    println!("B4: {rounds} budget checks over {N} entries — scan {scan:?}, aggregate {aggregate:?}");
+    println!(
+        "B4: {rounds} budget checks over {N} entries — scan {scan:?}, aggregate {aggregate:?}"
+    );
 }
 
 /// Victims still come out least-recently-accessed first, now in bounded batches

crates/pdfs-fuse/src/control.rs

@@ -165,43 +165,54 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
         Ok(CtlRequest::UploadPhoto {
             name,
             media_type,
-            bytes,
+            source_path,
             capture_time,
         }) => {
-            let photos = core.photos();
+            // Streamed off disk rather than carried over the socket: see the note
-            let metadata = proton_drive_rs::PhotoUploadMetadata {
+            // on `Request::UploadPhoto`. The length comes from the same handle the
-                capture_time,
+            // bytes do, so a file changing size under us cannot make the two
-                ..Default::default()
+            // disagree.
-            };
+            match std::fs::File::open(&source_path).and_then(|f| f.metadata().map(|m| (f, m.len())))
-            let guard = core.transfers.begin(
+            {
-                name.clone(),
+                Err(e) => CtlResponse::error(CoreError::invalid(format!(
-                "",
+                    "cannot read {source_path}: {e}"
-                TransferDirection::Upload,
+                ))),
-                bytes.len() as u64,
+                Ok((file, len)) => {
-            );
+                    let photos = core.photos();
-            let reader = CountingReader::new(std::io::Cursor::new(&bytes), &guard);
+                    let metadata = proton_drive_rs::PhotoUploadMetadata {
-            match core.rt.block_on(photos.upload_photo_from(
+                        capture_time,
-                &name,
+                        ..Default::default()
-                &media_type,
+                    };
-                reader,
+                    let guard =
-                bytes.len() as i64,
+                        core.transfers
-                Vec::new(),
+                            .begin(name.clone(), "", TransferDirection::Upload, len);
-                metadata,
+                    let reader = CountingReader::new(std::io::BufReader::new(file), &guard);
-                false,
+                    match core.rt.block_on(photos.upload_photo_from(
-            )) {
+                        &name,
-                Ok(uid) => {
+                        &media_type,
-                    // The photo we just uploaded belongs at the head of the
+                        reader,
-                    // timeline, and the GUI reloads the gallery the moment this
+                        len as i64,
-                    // reply lands — so refresh now rather than leaving it to a
+                        Vec::new(),
-                    // background pass that would land just after that reload.
+                        metadata,
-                    if let Err(e) = core.rt.block_on(core.refresh_timeline()) {
+                        false,
-                        warn!(error = %e, "timeline refresh after upload failed");
+                    )) {
-                    }
+                        Ok(uid) => {
-                    CtlResponse::Ok {
+                            // The photo we just uploaded belongs at the head of the
-                        message: format!("uploaded photo with uid {uid}"),
+                            // timeline, and the GUI reloads the gallery the moment this
+                            // reply lands — so refresh now rather than leaving it to a
+                            // background pass that would land just after that reload.
+                            if let Err(e) = core.rt.block_on(core.refresh_timeline()) {
+                                warn!(error = %e, "timeline refresh after upload failed");
+                            }
+                            CtlResponse::Ok {
+                                message: format!("uploaded photo with uid {uid}"),
+                            }
+                        }
+                        Err(e) => {
+                            CtlResponse::error(CoreError::from_api(&e, "upload photo failed"))
+                        }
                     }
                 }
-                Err(e) => CtlResponse::error(CoreError::from_api(&e, "upload photo failed")),
             }
         }
         Ok(CtlRequest::OpenFile { path }) => match rel_to_mount(mountpoint, &path) {
@@ -294,25 +305,6 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             },
             Err(e) => CtlResponse::error(e),
         },
-        Ok(CtlRequest::UploadFile {
-            parent,
-            name,
-            bytes,
-        }) => match rel_to_mount(mountpoint, &parent) {
-            Ok(parent_rel) => match core.upload(&parent_rel, &name, &bytes) {
-                Ok(name) => {
-                    core.log_activity(ActivityKind::Upload, &name, "", true);
-                    CtlResponse::Ok {
-                        message: format!("uploaded {name}"),
-                    }
-                }
-                Err(e) => {
-                    core.log_activity(ActivityKind::Upload, &name, &e, false);
-                    CtlResponse::error(e)
-                }
-            },
-            Err(e) => CtlResponse::error(e),
-        },
         Ok(CtlRequest::UploadPaths { parent, sources }) => {
             match rel_to_mount(mountpoint, &parent) {
                 // Ack immediately and upload on a background thread: a directory tree

crates/pdfs-fuse/src/devices.rs

@@ -555,24 +555,44 @@ impl Core {
                     .next()
                     .ok_or_else(|| SwitchBlocked::Failed("remote folder not found".to_string()))?;
 
-                // Reclaim the disk: empty the local dir (keep it as the mountpoint).
+                // Persist the mode *before* touching the local tree, and never roll it
-                evict_dir_contents(&local).map_err(|e| {
+                // back on failure.
-                    SwitchBlocked::Failed(format!("evict {}: {e}", local.display()))
+                //
-                })?;
+                // The eviction below deletes every local file in the folder. While the
-
+                // row still says `mirror`, that empty directory is a valid input to the
-                let session = self
+                // sync engine, which reads it as "the user deleted all of this" and
-                    .spawn_ondemand_mount(&local, root)
+                // trashes the whole folder on Drive. So the window between the evict and
-                    .map_err(|e| SwitchBlocked::Failed(e.to_string()))?;
+                // the mode write is a window in which any failure — a busy mountpoint, a
-                self.mounts.lock().insert(id, session);
+                // missing fuse module, a dead DB — destroys the folder. Writing the mode
-                // Persist the mode only now that the mount is actually up. Writing it
+                // first closes it: the engine skips non-mirror folders, so the worst a
-                // first would strand the folder on any failure below: the engine skips
+                // failure can now leave behind is an `ondemand` row with no mount, which
-                // non-mirror folders, so an `ondemand` row with no mount is a folder
+                // is inert. Inert and marked `error` is recoverable; trashed is not.
-                // that is neither mirrored nor browsable, and nothing retries it.
+                //
-                // Failing before this point leaves it `mirror`, and the next pass
+                // The same reasoning forbids reverting to `mirror` when a step below
-                // re-downloads whatever eviction removed.
+                // fails, since eviction may already have deleted part of the tree.
+                // Recovery is the user's explicit switch back to `mirror`, whose arm
+                // clears the baseline and re-downloads.
                 self.db
                     .sync_folder_set_mode(id, "ondemand")
                     .map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?;
+
+                // Reclaim the disk: empty the local dir (keep it as the mountpoint).
+                if let Err(e) = evict_dir_contents(&local) {
+                    let _ = self.db.sync_folder_set_state(id, "error", now_secs());
+                    return Err(SwitchBlocked::Failed(format!(
+                        "evict {}: {e}",
+                        local.display()
+                    )));
+                }
+
+                let session = match self.spawn_ondemand_mount(&local, root) {
+                    Ok(session) => session,
+                    Err(e) => {
+                        let _ = self.db.sync_folder_set_state(id, "error", now_secs());
+                        return Err(SwitchBlocked::Failed(e.to_string()));
+                    }
+                };
+                self.mounts.lock().insert(id, session);
                 let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
                 self.db.sync_folder_set_state(id, "idle", now_secs()).ok();
                 info!(id, path = %local.display(), "mounted sync folder on-demand");

crates/pdfs-fuse/src/drain.rs

@@ -731,10 +731,11 @@ impl Core {
 
         // The staged blob now matches the sealed revision, so a pinned file keeps
         // it as its cached content rather than re-downloading what we just sent.
-        if self.cache.is_pinned(&uid)
+        // Adopted by link, not by value: the blob is the whole file, and a pinned
-            && let Ok(bytes) = std::fs::read(&blob)
+        // video read into a `Vec` here is an OOM. `discard_staged` below drops the
-        {
+        // staging name only — the cache keeps the inode.
-            let _ = self.cache.store(&uid, now_secs(), meta.len, &bytes);
+        if self.cache.is_pinned(&uid) {
+            let _ = self.cache.store_file(&uid, now_secs(), meta.len, &blob);
         }
         self.cache.discard_staged(&blob);
         self.evict_reader(&uid);

crates/pdfs-fuse/src/lib.rs

@@ -81,8 +81,8 @@ mod transfers;
 mod workers;
 use state::{Entry, Intervals, PendingRevision, State, WriteHandle};
 use tracing::{debug, error, info, warn};
-use transfers::{CountingReader, CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry};
+use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry};
-use workers::{FUSE_WORKERS, Workers};
+use workers::{FUSE_WORKERS, Lane, Workers};
 
 /// Attribute/entry cache lifetime handed back to the kernel. Long because the
 /// Phase 2 event poller actively invalidates changed inodes; without a remote
@@ -637,6 +637,67 @@ impl Core {
         }
     }
 
+    /// Queue the writes that an unclean shutdown caught between `fsync(2)` and
+    /// `close(2)`.
+    ///
+    /// `fsync` promises the bytes survive a crash, but the queueing that makes a
+    /// write outlive the daemon happens at `release`. A crash in between used to
+    /// lose the data outright, because the scratch directory is cleared at open.
+    /// Now `fsync` leaves a sidecar, `ContentCache::open` moves those blobs to
+    /// `recovery/`, and this walks them into the same staging + queued-op path a
+    /// normal release takes.
+    ///
+    /// Runs after [`hydrate_pending`](Self::hydrate_pending), which is what makes
+    /// the incomplete-blob check in [`enqueue_staged_write`](Self::enqueue_staged_write)
+    /// meaningful: a recovered partial write whose earlier write is still queued
+    /// must not gap-fill from a remote revision that no longer describes the
+    /// file. With `pending` already loaded, that case is detected and the bytes
+    /// are parked rather than mis-filled.
+    ///
+    /// Failure is per-write and never fatal: a node that no longer exists, or an
+    /// op that cannot be queued, leaves its blob in `recovery/` for the next run
+    /// (and for a human) instead of taking the mount down.
+    fn recover_fsynced_writes(&self) {
+        let recovered = self.cache.recovered_writes();
+        if recovered.is_empty() {
+            return;
+        }
+        let mut queued = 0usize;
+        for (blob, meta) in recovered {
+            let Some(uid) = parse_node_uid(&meta.uid) else {
+                error!(uid = %meta.uid, "recovered write has an unparseable uid; keeping bytes");
+                continue;
+            };
+            // The inode is only used to stamp the in-memory tree with the new
+            // size, and nothing is interned this early — `hydrate` reads the
+            // size back off `pending` when the node is first looked up, so 0
+            // (no such inode) is correct rather than merely tolerable.
+            match self.enqueue_staged_write(&uid, 0, &blob, meta) {
+                Ok(()) => {
+                    self.cache.discard_recovered(&blob);
+                    queued += 1;
+                }
+                Err(e) => {
+                    error!(%uid, blob = %blob.display(), error = ?e,
+                           "cannot queue a recovered write; bytes kept for the next run");
+                    // Some failures (a partial write parked by
+                    // `stage_orphaned_write`) still consume the blob. Its sidecar
+                    // would then describe nothing, so retire it — the bytes are
+                    // in staging, which is where a human looks for them.
+                    if !blob.exists() {
+                        self.cache.discard_recovered(&blob);
+                    }
+                }
+            }
+        }
+        if queued > 0 {
+            info!(
+                count = queued,
+                "recovered fsynced writes from an unclean shutdown"
+            );
+        }
+    }
+
     /// Discard an op that can never be performed, because the bytes it was to
     /// upload are gone from staging (something outside the daemon deleted them).
     ///
@@ -1392,6 +1453,11 @@ impl Core {
     /// The scratch file is *moved* into staging, never copied: it is the only
     /// copy of what the user wrote.
     fn queue_revision(&self, h: &WriteHandle) -> Result<(), Errno> {
+        // The handle is being retired either way, so any durability sidecar an
+        // `fsync` left has done its job: from here the bytes are tracked as a
+        // staged write and a queued op, and a sidecar outliving them would offer
+        // recovery a second, stale copy of the same write.
+        self.cache.clear_scratch_durable(&h.path);
         if !h.dirty {
             let _ = std::fs::remove_file(&h.path);
             return Ok(());
@@ -2423,46 +2489,6 @@ impl Core {
         Ok(name.to_string())
     }
 
-    /// Upload a file named `name` with content `bytes` into the
-    /// mountpoint-relative `parent_rel` folder. Interns the new node directly.
-    fn upload(&self, parent_rel: &Path, name: &str, bytes: &[u8]) -> CoreResult<String> {
-        if name.is_empty() || name.contains('/') {
-            return Err(CoreError::invalid(format!("invalid name: {name:?}")));
-        }
-        let (pino, parent_uid) = self.resolve(parent_rel)?;
-        self.ensure_children(pino)
-            .map_err(|e| self.errno_error(e, "enumerate"))?;
-        let guard = self
-            .transfers
-            .begin(name, "", TransferDirection::Upload, bytes.len() as u64);
-        let reader = CountingReader::new(std::io::Cursor::new(bytes), &guard);
-        let new_uid = self
-            .rt
-            .block_on(self.client.upload_file_from(
-                &parent_uid,
-                name,
-                media_type_for(name),
-                reader,
-                bytes.len() as i64,
-                Vec::new(),
-                None,
-                false,
-            ))
-            .map_err(|e| CoreError::from_api(&e, "upload"))?;
-        drop(guard);
-        let node = self
-            .fetch_node(&new_uid)
-            .map_err(|e| self.errno_error(e, "fetch node"))?;
-        let mut st = self.state.lock();
-        let ino = st.intern(pino, node);
-        if let Some(kids) = st.children.get_mut(&pino)
-            && !kids.contains(&ino)
-        {
-            kids.push(ino);
-        }
-        Ok(name.to_string())
-    }
-
     /// Bulk-upload local files and directory trees under `sources` into the
     /// mountpoint-relative `parent_rel` folder. Directories are recreated (or
     /// merged into an existing same-named folder) and walked; the resulting flat
@@ -3328,7 +3354,7 @@ impl Filesystem for ProtonFs {
         let fs = self.clone();
         self.core
             .workers
-            .run(move || fs.serve_lookup(parent, &name, reply));
+            .run(Lane::Meta, move || fs.serve_lookup(parent, &name, reply));
     }
 
     fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option<FileHandle>, reply: ReplyAttr) {
@@ -3360,7 +3386,7 @@ impl Filesystem for ProtonFs {
         let fs = self.clone();
         self.core
             .workers
-            .run(move || fs.serve_readdir(ino, offset, reply));
+            .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply));
     }
 
     fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
@@ -3503,7 +3529,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(move || {
+        self.core.workers.run(Lane::Transfer, move || {
             match core.read_range(&uid, mtime, fsize, offset, size as u64, cache_blocks) {
                 Ok(bytes) => reply.data(&bytes),
                 Err(e) => reply.error(e),
@@ -3793,23 +3819,67 @@ impl Filesystem for ProtonFs {
         _datasync: bool,
         reply: ReplyEmpty,
     ) {
-        let file = self
+        // Everything the durability sidecar needs, copied out so no lock is held
-            .core
+        // across the sync.
-            .state
+        let handle = self.core.state.lock().handles.get(&fh.0).map(|h| {
-            .lock()
+            (
-            .handles
+                h.file.clone(),
-            .get(&fh.0)
+                h.path.clone(),
-            .map(|h| h.file.clone());
+                h.uid.clone(),
-        match file {
+                h.dirty,
-            Some(f) => match f.sync_all() {
+                h.written.clone(),
-                Ok(()) => reply.ok(),
+                h.len,
-                Err(e) => {
+                h.base_size,
-                    error!(fh = fh.0, error = %e, "fsync of scratch file failed");
+                h.base_mtime,
-                    reply.error(Errno::EIO);
+            )
-                }
+        });
-            },
+        let Some((f, path, uid, dirty, written, len, base_size, base_mtime)) = handle else {
-            None => reply.error(Errno::EBADF),
+            reply.error(Errno::EBADF);
+            return;
+        };
+        if let Err(e) = f.sync_all() {
+            error!(fh = fh.0, error = %e, "fsync of scratch file failed");
+            reply.error(Errno::EIO);
+            return;
         }
+        // A clean handle has nothing on it that the remote does not already
+        // hold, so there is nothing for a crash to lose.
+        if !dirty {
+            reply.ok();
+            return;
+        }
+        // The bytes are on stable storage, but only the scratch file knows that,
+        // and the scratch directory is cleared at open. Record what the blob is
+        // so a restart can hand it to the upload queue instead of deleting it.
+        //
+        // Deliberately *not* the full release path: staging the blob here would
+        // move the file out from under a handle the application still has open,
+        // and queueing an upload per `fsync` would push a revision for every
+        // barrier in a write loop. This only makes the existing file findable.
+        let authored: Vec<(u64, u64)> = written
+            .segments(0, len)
+            .into_iter()
+            .filter(|&(_, _, authored)| authored)
+            .map(|(s, e, _)| (s, e))
+            .collect();
+        let meta = StagedWrite {
+            uid: uid.to_string(),
+            len,
+            base_size,
+            base_mtime,
+            complete: authored == [(0, len)],
+            authored,
+            based_on: self.core.remote_baseline(&uid, base_mtime, base_size),
+        };
+        // A failed sidecar means the write is not durable, and `fsync` promising
+        // otherwise is the defect this exists to fix — so it is an error, not a
+        // warning.
+        if let Err(e) = self.core.cache.mark_scratch_durable(&path, &meta) {
+            error!(%uid, error = %e, "recording fsync durability failed");
+            reply.error(Errno::EIO);
+            return;
+        }
+        reply.ok();
     }
 
     fn release(
@@ -4037,17 +4107,10 @@ impl Filesystem for ProtonFs {
             reply.error(Errno::EIO);
             return;
         }
-        // Forget the node so it re-interns under its new parent, and drop the
+        self.core
-        // destination listing so it re-enumerates on next access.
+            .state
-        //
+            .lock()
-        // The drop has to go through `invalidate_listing`, not `children.remove`:
+            .relocate(ino, parent, newparent, &new_parent_uid, &newname);
-        // `forget` deletes the node's DB row, so a destination still marked
-        // `listed` would be rebuilt from the DB by `ensure_children` — without the
-        // node we just moved into it. The file would be gone from the source and
-        // absent from the destination, with the rename having reported success.
-        let mut st = self.core.state.lock();
-        st.forget(&uid);
-        st.invalidate_listing(newparent);
         reply.ok();
     }
 
@@ -4526,6 +4589,10 @@ pub fn mount(
     // Writes queued by a previous run (or left behind by a crash) are still owed
     // an upload, and reads must be served from their staged blobs until they land.
     core.hydrate_pending();
+    // Then the writes that were fsynced but never closed, which the cache moved
+    // aside at open. After `hydrate_pending`, so a recovered partial write can
+    // see an already-queued write to the same node.
+    core.recover_fsynced_writes();
     {
         let core = core.clone();
         std::thread::Builder::new()
@@ -4584,6 +4651,15 @@ pub fn mount(
     // stale socket file from a previous run would block the bind, so clear it.
     let _ = std::fs::remove_file(control_socket);
     let listener = UnixListener::bind(control_socket)?;
+    // Owner-only before anything can connect: a peer on this socket commands the
+    // daemon's authenticated session without a credential of its own (B6).
+    if let Err(e) = pdfs_core::config::restrict_socket(control_socket) {
+        error!(error = %e, "cannot restrict control socket permissions; refusing to serve");
+        let _ = std::fs::remove_file(control_socket);
+        return Err(std::io::Error::other(format!(
+            "control socket permissions: {e}"
+        )));
+    }
     {
         let core = core.clone();
         let username = username.clone();

crates/pdfs-fuse/src/state.rs

@@ -269,6 +269,43 @@ impl State {
         }
     }
 
+    /// Settle the local state after a node has been moved (and possibly renamed)
+    /// on the remote: rewrite it in place, and drop **both** directories'
+    /// listings so each re-enumerates.
+    ///
+    /// Both, because each is stale for its own reason. The destination has
+    /// gained a child it does not know about; the source has lost one. The
+    /// source is the subtler half, and was audit A5: pruning the moved node from
+    /// the source's in-memory children looks sufficient while that entry stays
+    /// resident, but the source's DB row is left `listed = 1`. Once it is
+    /// evicted from the hot cache, or the daemon restarts, `ensure_children`
+    /// rebuilds the listing from the DB and declares it complete, so anything
+    /// else that changed remotely under the source since is never seen.
+    ///
+    /// What this must *not* do is `forget` the node. Forgetting drops its
+    /// `by_uid` mapping, so the re-enumeration hands it a fresh inode — while
+    /// the kernel has already carried the renamed dentry over to the *old* one.
+    /// Every lookup through that dentry then resolves to an inode `entries` no
+    /// longer holds and fails `ENOENT`, so the renamed directory reads as
+    /// missing (`ls` on it errors while `ls` of its parent lists it) until the
+    /// entry TTL expires. Keeping the inode keeps the kernel's dentry valid.
+    ///
+    /// A pure rename (`from == to`) invalidates that one directory once.
+    pub(crate) fn relocate(
+        &mut self,
+        ino: u64,
+        from_parent: u64,
+        to_parent: u64,
+        to_parent_uid: &NodeUid,
+        name: &str,
+    ) {
+        self.rename_in_place(ino, to_parent, to_parent_uid, name);
+        self.invalidate_listing(to_parent);
+        if from_parent != to_parent {
+            self.invalidate_listing(from_parent);
+        }
+    }
+
     /// Update a file entry's recorded plaintext size so `getattr` reflects an
     /// in-progress write before the new revision is sealed.
     pub(crate) fn set_size(&mut self, ino: u64, size: u64) {
@@ -305,3 +342,154 @@ impl State {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
+
+    fn uid(link: &str) -> NodeUid {
+        NodeUid::new(VolumeId::from("vol"), LinkId::from(link))
+    }
+
+    fn node(link: &str, parent: &str, name: &str, is_dir: bool) -> Node {
+        Node {
+            uid: uid(link),
+            parent_uid: Some(uid(parent)),
+            kind: if is_dir {
+                NodeKind::Folder
+            } else {
+                NodeKind::File {
+                    media_type: "text/plain".into(),
+                    total_size_on_storage: 0,
+                    active_revision_state: None,
+                    claimed_size: Some(0),
+                    claimed_modification_time: None,
+                }
+            },
+            name: name.into(),
+            creation_time: 100,
+            modification_time: 100,
+            trashed: false,
+            is_shared: false,
+            is_shared_publicly: false,
+            signature_email: None,
+            verification: Default::default(),
+        }
+    }
+
+    /// A unique temp directory removed on drop; avoids a dev-dependency.
+    struct TempDir(PathBuf);
+    impl TempDir {
+        fn new() -> Self {
+            use std::sync::atomic::{AtomicU64, Ordering};
+            static N: AtomicU64 = AtomicU64::new(0);
+            let p = std::env::temp_dir().join(format!(
+                "pdfs-state-{}-{}",
+                std::process::id(),
+                N.fetch_add(1, Ordering::Relaxed)
+            ));
+            std::fs::create_dir_all(&p).unwrap();
+            TempDir(p)
+        }
+    }
+    impl Drop for TempDir {
+        fn drop(&mut self) {
+            let _ = std::fs::remove_dir_all(&self.0);
+        }
+    }
+
+    /// `Db::open_in_memory` is `#[cfg(test)]` inside `pdfs-core`, so it does not
+    /// exist for this crate's tests — a temp file is the equivalent here. The
+    /// directory outlives the state because the state holds the open database.
+    fn state() -> (State, TempDir) {
+        let dir = TempDir::new();
+        let db = Db::open(&dir.0.join("cache.db")).unwrap();
+        let st = State {
+            entries: HashMap::new(),
+            by_uid: HashMap::new(),
+            children: HashMap::new(),
+            next_ino: 1,
+            handles: HashMap::new(),
+            next_fh: 1,
+            db: Arc::new(db),
+        };
+        (st, dir)
+    }
+
+    /// Two folders, each holding one file, both listed — the state a move
+    /// starts from. Returns `(state, src_ino, dst_ino)`.
+    fn two_folders() -> (State, TempDir, u64, u64) {
+        let (mut st, dir) = state();
+        let src = st.intern(0, node("src", "root", "src", true));
+        let dst = st.intern(0, node("dst", "root", "dst", true));
+        let f = st.intern(src, node("f", "src", "f.txt", false));
+        st.children.insert(src, vec![f]);
+        st.children.insert(dst, vec![]);
+        st.db.set_listed(&uid("src"), true).unwrap();
+        st.db.set_listed(&uid("dst"), true).unwrap();
+        (st, dir, src, dst)
+    }
+
+    /// Audit A5. Moving a file out of a folder leaves that folder's DB row
+    /// claiming a complete listing, so once the hot cache drops it, remote
+    /// changes under it stop being seen. `relocate` has to clear both sides.
+    #[test]
+    fn relocate_invalidates_the_source_as_well_as_the_destination() {
+        let (mut st, _dir, src, dst) = two_folders();
+        let f = st.by_uid[&uid("f")];
+        st.relocate(f, src, dst, &uid("dst"), "f.txt");
+
+        assert!(
+            st.db.children_if_listed(&uid("dst")).unwrap().is_none(),
+            "destination re-enumerates: it gained a child whose row was just deleted"
+        );
+        assert!(
+            st.db.children_if_listed(&uid("src")).unwrap().is_none(),
+            "source re-enumerates: its listing predates the move, and a stale \
+             `listed` flag would hide every later remote change under it"
+        );
+        assert!(!st.children.contains_key(&src));
+        assert!(!st.children.contains_key(&dst));
+        assert_eq!(
+            st.by_uid.get(&uid("f")),
+            Some(&f),
+            "the inode survives the move: the kernel has already pointed the \
+             renamed dentry at it, so re-interning under a fresh one would make \
+             every lookup through that dentry ENOENT"
+        );
+        assert_eq!(st.entries[&f].parent, dst);
+    }
+
+    /// A pure rename never leaves the directory, so there is one listing to
+    /// drop, not two — and dropping it must not depend on the parents differing.
+    #[test]
+    fn relocate_within_one_directory_still_invalidates_it() {
+        let (mut st, _dir, src, _dst) = two_folders();
+        let f = st.by_uid[&uid("f")];
+        st.relocate(f, src, src, &uid("src"), "renamed.txt");
+        assert!(st.db.children_if_listed(&uid("src")).unwrap().is_none());
+        assert!(!st.children.contains_key(&src));
+        assert_eq!(st.by_uid.get(&uid("f")), Some(&f), "the inode is stable");
+        assert_eq!(st.entries[&f].node.name, "renamed.txt");
+    }
+
+    /// The failure mode the fix exists for, stated directly: forgetting the node
+    /// prunes it from the resident listing but leaves the DB claiming the source
+    /// is fully enumerated. This pins *why* `relocate` cannot just call `forget`.
+    #[test]
+    fn forget_alone_leaves_the_source_claiming_a_complete_listing() {
+        let (mut st, _dir, _src, _dst) = two_folders();
+        st.forget(&uid("f"));
+        let listed = st.db.children_if_listed(&uid("src")).unwrap();
+        assert!(
+            listed.is_some(),
+            "forget does not clear the flag — which is exactly why relocate must"
+        );
+        assert!(
+            listed.unwrap().is_empty(),
+            "and the listing it would serve is the moved file's absence, \
+             with no way to notice anything else changed"
+        );
+    }
+}

crates/pdfs-fuse/src/sync.rs

@@ -17,8 +17,8 @@ use std::collections::{HashMap, HashSet};
 use std::io::Write;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use std::sync::mpsc::Receiver;
+use std::sync::mpsc::{Receiver, RecvTimeoutError};
-use std::time::{Duration, SystemTime};
+use std::time::{Duration, Instant, SystemTime};
 
 use notify::{RecursiveMode, Watcher};
 use proton_drive_rs::proton_sdk::ids::NodeUid;
@@ -30,8 +30,20 @@ use crate::{Core, now_secs, parse_uid};
 use pdfs_core::control::{ActivityKind, TransferDirection};
 use pdfs_core::db::{StoredSyncEntry, StoredSyncFolder};
 
-/// Coalesce a burst of filesystem events into a single reconcile per folder.
+/// How long the watcher must go quiet before a burst counts as finished.
+///
+/// Measured from the *last* event, not the first: applications do not write a
+/// file once. An editor saves by writing a temp file and renaming it over the
+/// target; an export or a database dump writes continuously for as long as it
+/// takes. Settling a fixed interval after the first event walks the tree while
+/// the file is still growing, and uploads a torn snapshot as a real revision —
+/// corrected on the next pass, but only after spending the encrypt-and-upload
+/// on a file that was never in that state.
 const DEBOUNCE: Duration = Duration::from_secs(2);
+/// Ceiling on the settle wait, so a directory under sustained change still
+/// syncs. Without it, copying a large tree in — events arriving forever —
+/// would postpone the reconcile for as long as the copy ran.
+const MAX_SETTLE: Duration = Duration::from_secs(30);
 /// How often to re-walk remotes so changes made on other clients are pulled in.
 const POLL_INTERVAL: Duration = Duration::from_secs(120);
 /// How many uploads/downloads/folder-creations a reconcile runs at once. The
@@ -111,11 +123,7 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
         let mut all = false;
         let mut do_rewatch = false;
         classify(msg, &mut ids, &mut all, &mut do_rewatch);
-        // Debounce: let a burst settle, then drain everything queued behind it.
+        settle(&rx, &mut ids, &mut all, &mut do_rewatch);
-        std::thread::sleep(DEBOUNCE);
-        while let Ok(m) = rx.try_recv() {
-            classify(m, &mut ids, &mut all, &mut do_rewatch);
-        }
 
         if do_rewatch {
             rewatch(&core, &mut watcher, &watched);
@@ -151,10 +159,7 @@ fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
         let mut all = false;
         let mut do_rewatch = false;
         classify(msg, &mut ids, &mut all, &mut do_rewatch);
-        std::thread::sleep(DEBOUNCE);
+        settle(&rx, &mut ids, &mut all, &mut do_rewatch);
-        while let Ok(m) = rx.try_recv() {
-            classify(m, &mut ids, &mut all, &mut do_rewatch);
-        }
         if all {
             reconcile_all(core);
         } else {
@@ -167,6 +172,46 @@ fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
     }
 }
 
+/// Absorb the rest of an event burst, returning once the watcher has been quiet
+/// for `quiet` or `cap` has elapsed since the first event — whichever comes
+/// first. The caller has already classified the event that opened the burst.
+///
+/// Split out with its timings as parameters so the settling behaviour can be
+/// tested in milliseconds rather than in the tens of seconds the real constants
+/// describe.
+fn settle_with(
+    rx: &Receiver<SyncMsg>,
+    ids: &mut HashSet<i64>,
+    all: &mut bool,
+    do_rewatch: &mut bool,
+    quiet: Duration,
+    cap: Duration,
+) {
+    let deadline = Instant::now() + cap;
+    loop {
+        let now = Instant::now();
+        if now >= deadline {
+            break;
+        }
+        // Never wait past the cap, however quiet the watcher goes.
+        let wait = quiet.min(deadline - now);
+        match rx.recv_timeout(wait) {
+            // Another event: the burst is still going, so the quiet window
+            // starts again from here.
+            Ok(m) => classify(m, ids, all, do_rewatch),
+            // Quiet for a full window: whatever was being written has stopped.
+            Err(RecvTimeoutError::Timeout) => break,
+            // Every sender is gone; the caller's loop is about to end anyway.
+            Err(RecvTimeoutError::Disconnected) => break,
+        }
+    }
+}
+
+/// [`settle_with`] at the real timings.
+fn settle(rx: &Receiver<SyncMsg>, ids: &mut HashSet<i64>, all: &mut bool, do_rewatch: &mut bool) {
+    settle_with(rx, ids, all, do_rewatch, DEBOUNCE, MAX_SETTLE);
+}
+
 fn classify(msg: SyncMsg, ids: &mut HashSet<i64>, all: &mut bool, rewatch: &mut bool) {
     match msg {
         SyncMsg::Reconcile(id) => {
@@ -386,6 +431,37 @@ impl From<String> for PassAbort {
     }
 }
 
+/// Refuse to run a pass whose local side has vanished in its entirety.
+///
+/// A path in the baseline with no local copy is classified as "the user deleted
+/// this", and the pass answers by trashing it on Drive. That is right for one
+/// file and right for a handful. When it is true of *every* path the baseline
+/// knows, the likely cause is not a user: it is an unmounted external disk, a
+/// mountpoint whose FUSE mount died, a home directory that was not yet decrypted
+/// when the daemon started, or a bug in this client — and the answer to all of
+/// those is to trash the folder's entire contents on Drive.
+///
+/// So this pass abort exists to cap the blast radius rather than to catch a
+/// specific defect. Losing a pass to it costs a retry; the failure it prevents
+/// costs the folder. A user who really did empty the folder can express that by
+/// removing it from sync and re-adding it — the baseline goes with it.
+///
+/// One surviving path is enough to clear the guard, since the classification is
+/// only ever wrong about all of them at once.
+fn guard_local_wipe<B, L>(
+    baseline: &HashMap<String, B>,
+    local: &HashMap<String, L>,
+) -> Result<(), String> {
+    if baseline.len() >= 2 && baseline.keys().all(|rel| !local.contains_key(rel)) {
+        return Err(format!(
+            "every one of the {} synced paths is missing locally; refusing to trash \
+             them on Drive. Check that the folder is mounted and readable.",
+            baseline.len()
+        ));
+    }
+    Ok(())
+}
+
 /// What a [`Pending`] op did, so the reconcile can update shared state on the
 /// engine thread (never inside a task).
 enum Applied {
@@ -592,6 +668,7 @@ impl Core {
 
         let mut local: HashMap<String, LocalItem> = HashMap::new();
         self.walk_local(folder_id, local_root, local_root, &mut local)?;
+        guard_local_wipe(&baseline, &local)?;
 
         // The remote folder uids come from the baseline instead of a walk: every
         // directory this device has synced recorded its uid there when it was
@@ -846,6 +923,7 @@ impl Core {
 
         let mut local: HashMap<String, LocalItem> = HashMap::new();
         self.walk_local(folder_id, local_root, local_root, &mut local)?;
+        guard_local_wipe(&baseline, &local)?;
 
         let mut remote: HashMap<String, RemoteItem> = HashMap::new();
         let mut remote_dirs: HashMap<String, NodeUid> = HashMap::new();
@@ -1741,6 +1819,113 @@ fn conflict_path(path: &Path, stamp: i64) -> PathBuf {
 mod tests {
     use super::*;
 
+    /// Feed `n` `Reconcile` events `gap` apart on another thread, starting
+    /// immediately. Returns the sender's join handle so the test can keep it
+    /// alive (dropping the sender early would look like a disconnect).
+    fn drip(n: usize, gap: Duration) -> (Receiver<SyncMsg>, std::thread::JoinHandle<()>) {
+        let (tx, rx) = std::sync::mpsc::channel();
+        let h = std::thread::spawn(move || {
+            for i in 0..n {
+                if tx.send(SyncMsg::Reconcile(i as i64)).is_err() {
+                    return;
+                }
+                std::thread::sleep(gap);
+            }
+        });
+        (rx, h)
+    }
+
+    fn collect(rx: &Receiver<SyncMsg>, quiet: Duration, cap: Duration) -> (HashSet<i64>, Duration) {
+        let mut ids = HashSet::new();
+        let (mut all, mut rewatch) = (false, false);
+        // The caller of `settle` has always classified the opening event first.
+        let first = rx.recv().unwrap();
+        classify(first, &mut ids, &mut all, &mut rewatch);
+        let started = Instant::now();
+        settle_with(rx, &mut ids, &mut all, &mut rewatch, quiet, cap);
+        (ids, started.elapsed())
+    }
+
+    /// The bug this replaced a fixed sleep for: a save that keeps writing past
+    /// the debounce used to be walked mid-write, uploading a torn snapshot as a
+    /// real revision. The window has to restart on every event, so settling
+    /// waits for the writer to actually stop.
+    #[test]
+    fn settling_waits_for_the_last_event_not_the_first() {
+        // Eight events 20ms apart — a burst lasting ~160ms, far longer than the
+        // 60ms quiet window a fixed sleep would have used.
+        let (rx, h) = drip(8, Duration::from_millis(20));
+        let (ids, elapsed) = collect(&rx, Duration::from_millis(60), Duration::from_secs(30));
+        h.join().unwrap();
+
+        assert_eq!(ids.len(), 8, "every event in the burst is absorbed");
+        assert!(
+            elapsed >= Duration::from_millis(120),
+            "settling returned after {elapsed:?}: it stopped waiting while events \
+             were still arriving, which is the torn-revision bug"
+        );
+    }
+
+    /// A burst that ends is not waited on any longer than the quiet window.
+    #[test]
+    fn settling_returns_once_the_burst_stops() {
+        let (rx, h) = drip(3, Duration::from_millis(5));
+        let (ids, elapsed) = collect(&rx, Duration::from_millis(50), Duration::from_secs(30));
+        h.join().unwrap();
+
+        assert_eq!(ids.len(), 3);
+        assert!(
+            elapsed < Duration::from_secs(5),
+            "a finished burst must settle on the quiet window, not the cap"
+        );
+    }
+
+    /// Sustained change — copying a large tree in — must not postpone the
+    /// reconcile for as long as the copy runs. The cap ends the wait.
+    #[test]
+    fn settling_is_capped_under_continuous_change() {
+        // Events every 10ms for far longer than the 150ms cap allows.
+        let (rx, h) = drip(200, Duration::from_millis(10));
+        let (_ids, elapsed) = collect(&rx, Duration::from_millis(100), Duration::from_millis(150));
+        assert!(
+            elapsed < Duration::from_secs(2),
+            "settled after {elapsed:?}; the cap should have ended the wait near 150ms"
+        );
+        drop(rx);
+        h.join().unwrap();
+    }
+
+    /// The guard that keeps a vanished local tree (dead mount, unplugged disk)
+    /// from being read as "the user deleted everything" and trashing the folder
+    /// on Drive.
+    #[test]
+    fn guard_local_wipe_blocks_only_a_total_disappearance() {
+        let base: HashMap<String, ()> = ["a.txt", "b.txt", "d/c.txt"]
+            .iter()
+            .map(|r| ((*r).to_string(), ()))
+            .collect();
+        let none: HashMap<String, ()> = HashMap::new();
+        let one: HashMap<String, ()> = [("b.txt".to_string(), ())].into_iter().collect();
+
+        assert!(
+            guard_local_wipe(&base, &none).is_err(),
+            "wholesale loss aborts"
+        );
+        assert!(
+            guard_local_wipe(&base, &one).is_ok(),
+            "one survivor means the tree is readable; the rest are real deletions"
+        );
+        // A one-entry baseline is a genuine single delete, not a wipe to detect.
+        let single: HashMap<String, ()> = [("a.txt".to_string(), ())].into_iter().collect();
+        assert!(guard_local_wipe(&single, &none).is_ok());
+        // Nothing synced yet: a first pass over an empty folder is not a wipe.
+        assert!(guard_local_wipe(&none, &none).is_ok());
+        // Local paths that are not in the baseline are new uploads, and do not
+        // count as survivors.
+        let fresh: HashMap<String, ()> = [("new.txt".to_string(), ())].into_iter().collect();
+        assert!(guard_local_wipe(&base, &fresh).is_err());
+    }
+
     #[test]
     fn parent_and_base_split_relative_paths() {
         assert_eq!(parent_rel("a"), "");

crates/pdfs-fuse/src/workers.rs

@@ -3,10 +3,17 @@
 //! fuser drives `Filesystem` from a single dispatch loop, so a cold read that
 //! goes to the wire would block every cheap metadata call behind it. Those
 //! handlers hand their work here instead and reply from the worker.
+//!
+//! The pool is split into two lanes because moving the work off the dispatch
+//! loop is not by itself enough: with one shared queue, eight concurrent block
+//! downloads occupy every worker, and a `lookup` that needs one network round
+//! trip waits behind megabytes of transfer (audit A6). Some threads are
+//! therefore reserved for metadata and never accept a transfer.
 
+use std::collections::VecDeque;
 use std::sync::Arc;
 
-use parking_lot::Mutex;
+use parking_lot::{Condvar, Mutex};
 use tracing::warn;
 
 /// How many FUSE handlers may run off the dispatch loop at once.
@@ -19,75 +26,272 @@ use tracing::warn;
 ///
 /// Bounded on purpose: one worker can hold a 4 MiB block in flight, and an
 /// unbounded pool would let read-ahead on a big file spawn threads without
-/// limit. Sized just above the SDK's in-flight block cap so the semaphore
+/// limit. Sized so the SDK's in-flight block semaphore, not thread count, is
-/// there, not thread count, is what bounds download memory.
+/// what bounds download memory.
-pub(crate) const FUSE_WORKERS: usize = 8;
+///
+/// Counted as `META_WORKERS` reserved threads *plus* the eight that transfers
+/// had before the lanes were split. Taking the reservation out of the original
+/// eight instead would have cut concurrent reads to five — a throughput
+/// regression smuggled in with a latency fix. Threads are the cheap resource
+/// here; block buffers are the expensive one, and those are capped in the SDK.
+pub(crate) const FUSE_WORKERS: usize = 11;
+
+/// How many of [`FUSE_WORKERS`] serve metadata *only*.
+///
+/// This is the whole guarantee: these threads never accept a [`Lane::Transfer`]
+/// job, so no number of concurrent downloads can leave a `lookup` or `readdir`
+/// without a thread to run on. Three is enough because metadata jobs are short
+/// — one round trip, no block fetch — so they queue behind each other briefly
+/// rather than for the length of a transfer.
+///
+/// The remaining workers are general: they prefer transfers and fall back to
+/// metadata when there is no transfer waiting. That direction is safe (a
+/// general thread picking up a cheap job frees itself again quickly) while the
+/// reverse — metadata threads helping with transfers — would reintroduce
+/// exactly the blocking this split exists to prevent.
+const META_WORKERS: usize = 3;
+
+type Job = Box<dyn FnOnce() + Send + 'static>;
+
+/// Which lane a job belongs in.
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub(crate) enum Lane {
+    /// Short, network-bound metadata work: `lookup`, `readdir`. Guaranteed a
+    /// thread by the reserved workers.
+    Meta,
+    /// Bulk data movement: block reads. May occupy every general worker.
+    Transfer,
+}
+
+#[derive(Default)]
+struct Queues {
+    meta: VecDeque<Job>,
+    transfer: VecDeque<Job>,
+    /// Set when the pool is dropped, to wake and retire every worker.
+    closed: bool,
+}
+
+/// The queues and the two wait sets over them.
+///
+/// Two condvars, not one, because the workers are not interchangeable: a
+/// meta-only thread cannot take a transfer. With a single wait set,
+/// `notify_one` for a transfer could wake a meta-only thread, which would find
+/// nothing it is allowed to run and go back to sleep — leaving the transfer
+/// unclaimed while a general worker sat idle. Waking *every* thread instead
+/// would work but costs a thundering herd per job. So each lane notifies only
+/// threads that can actually serve it.
+struct Pool {
+    queues: Mutex<Queues>,
+    /// Waited on by the reserved metadata workers.
+    meta_cv: Condvar,
+    /// Waited on by the general workers.
+    general_cv: Condvar,
+}
 
 /// Bounded thread pool behind the network-touching FUSE handlers.
 ///
 /// Shared by every session forked off one [`Core`] (the main mount plus each
 /// on-demand sync folder), so the bound is per daemon rather than per mount.
 pub(crate) struct Workers {
-    pub(crate) tx: std::sync::mpsc::Sender<Box<dyn FnOnce() + Send + 'static>>,
+    inner: Arc<Pool>,
 }
 
 impl Workers {
     pub(crate) fn new(n: usize) -> std::io::Result<Self> {
-        let (tx, rx) = std::sync::mpsc::channel::<Box<dyn FnOnce() + Send + 'static>>();
+        let inner = Arc::new(Pool {
-        let rx = Arc::new(Mutex::new(rx));
+            queues: Mutex::new(Queues::default()),
+            meta_cv: Condvar::new(),
+            general_cv: Condvar::new(),
+        });
+        // Never leave the pool without a general worker, however small `n` is.
+        let meta_workers = META_WORKERS.min(n.saturating_sub(1));
         for i in 0..n {
-            let rx = rx.clone();
+            let inner = inner.clone();
+            let meta_only = i < meta_workers;
             std::thread::Builder::new()
-                .name(format!("pdfs-fuse-{i}"))
+                .name(format!(
+                    "pdfs-fuse-{}{i}",
+                    if meta_only { "meta-" } else { "" }
+                ))
                 .spawn(move || {
+                    let cv = if meta_only {
+                        &inner.meta_cv
+                    } else {
+                        &inner.general_cv
+                    };
                     loop {
-                        // Held only to take the next job, never across it.
+                        let mut q = inner.queues.lock();
-                        let job = rx.lock().recv();
+                        let job = loop {
-                        match job {
+                            // A general worker takes transfers first: metadata
-                            // A panicking handler must not cost the pool a
+                            // already has threads of its own, so draining the
-                            // worker for the rest of the run. The dropped
+                            // bulk queue is the useful thing for it to do.
-                            // `Reply` answers EIO on its own, so the caller of
+                            let picked = if meta_only {
-                            // the failed op is told; the next job is unaffected.
+                                q.meta.pop_front()
-                            //
+                            } else {
-                            // This only holds while no shared state a job
+                                q.transfer.pop_front().or_else(|| q.meta.pop_front())
-                            // touches sits behind a *poisoning* lock. A
+                            };
-                            // `std::sync::Mutex` held at the point of the panic
+                            if let Some(job) = picked {
-                            // comes back poisoned, and the worker we just
+                                break Some(job);
-                            // rescued dies on its next acquisition — as does
-                            // every other thread. The rescue would then convert
-                            // one recoverable EIO into a permanently broken
-                            // daemon: strictly worse than not catching at all.
-                            //
-                            // This crate's state is behind `parking_lot`, which
-                            // does not poison. **The requirement crosses the
-                            // crate boundary**, though: a job runs deep into
-                            // `proton-sdk`, whose entity cache is a `std`
-                            // mutex. It recovers the guard rather than
-                            // unwrapping (see `InMemoryCacheRepository::state`);
-                            // its session and HTTP state are
-                            // `tokio::sync::Mutex`, which has no poisoning at
-                            // all. Anything new reached from a job owes the same
-                            // check — an SDK-side `.lock().unwrap()` silently
-                            // invalidates this comment.
-                            Ok(job) => {
-                                let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job));
                             }
-                            // Every sender is gone: the daemon is shutting down.
+                            if q.closed {
-                            Err(_) => break,
+                                break None;
-                        }
+                            }
+                            cv.wait(&mut q);
+                        };
+                        drop(q);
+                        let Some(job) = job else { break };
+                        // 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
+                        // told; the next job is unaffected.
+                        //
+                        // This only holds while no shared state a job touches
+                        // sits behind a *poisoning* lock. A `std::sync::Mutex`
+                        // held at the point of the panic comes back poisoned,
+                        // and the worker we just rescued dies on its next
+                        // acquisition — as does every other thread. The rescue
+                        // would then convert one recoverable EIO into a
+                        // permanently broken daemon: strictly worse than not
+                        // catching at all.
+                        //
+                        // This crate's state is behind `parking_lot`, which
+                        // does not poison. **The requirement crosses the crate
+                        // boundary**, though: a job runs deep into
+                        // `proton-sdk`, whose entity cache is a `std` mutex. It
+                        // recovers the guard rather than unwrapping (see
+                        // `InMemoryCacheRepository::state`); its session and
+                        // HTTP state are `tokio::sync::Mutex`, which has no
+                        // 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));
                     }
                 })?;
         }
-        Ok(Self { tx })
+        Ok(Self { inner })
     }
 
-    /// Run `job` on a worker. Falls back to running it inline — the pre-pool
+    /// Queue `job` in `lane`.
-    /// behaviour — if no worker is left to take it, so a shut-down pool degrades
+    ///
-    /// to a slow mount rather than a mount that answers every read with EIO.
+    /// Neither queue is bounded, and deliberately so: the only thing that could
-    pub(crate) fn run(&self, job: impl FnOnce() + Send + 'static) {
+    /// apply backpressure here is the fuser dispatch loop, and blocking *that*
-        if let Err(e) = self.tx.send(Box::new(job)) {
+    /// is the stall this module exists to avoid. Queue depth is cheap anyway —
+    /// 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) {
+        let mut q = self.inner.queues.lock();
+        if q.closed {
+            // Pre-pool behaviour: a shut-down pool degrades to a slow mount
+            // rather than a mount that answers every read with EIO.
+            drop(q);
             warn!("fuse worker pool is gone; serving inline");
-            (e.0)();
+            job();
+            return;
+        }
+        match lane {
+            Lane::Meta => q.meta.push_back(Box::new(job)),
+            Lane::Transfer => q.transfer.push_back(Box::new(job)),
+        }
+        drop(q);
+        match lane {
+            // Either class can serve metadata, and only one of them needs to:
+            // whichever wakes first takes it, and the other finds the queue
+            // empty and waits again.
+            Lane::Meta => {
+                self.inner.meta_cv.notify_one();
+                self.inner.general_cv.notify_one();
+            }
+            // Only general workers may take a transfer, so waking a reserved
+            // one would be the lost wakeup this split is careful to avoid.
+            Lane::Transfer => {
+                self.inner.general_cv.notify_one();
+            }
         }
     }
 }
+
+impl Drop for Workers {
+    fn drop(&mut self) {
+        self.inner.queues.lock().closed = true;
+        self.inner.meta_cv.notify_all();
+        self.inner.general_cv.notify_all();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::mpsc;
+    use std::time::Duration;
+
+    /// Audit A6. Transfers filling every general worker must not delay a
+    /// metadata job: that is what the reserved threads are for.
+    #[test]
+    fn saturated_transfers_do_not_block_metadata() {
+        let pool = Workers::new(FUSE_WORKERS).unwrap();
+        let general = FUSE_WORKERS - META_WORKERS;
+
+        // Occupy every general worker with a transfer that will not finish
+        // until we say so, and wait until they are all actually running.
+        let (release_tx, release_rx) = mpsc::channel::<()>();
+        let release_rx = Arc::new(Mutex::new(release_rx));
+        let (started_tx, started_rx) = mpsc::channel();
+        for _ in 0..general {
+            let release_rx = release_rx.clone();
+            let started_tx = started_tx.clone();
+            pool.run(Lane::Transfer, move || {
+                started_tx.send(()).unwrap();
+                let _ = release_rx.lock().recv();
+            });
+        }
+        for _ in 0..general {
+            started_rx
+                .recv_timeout(Duration::from_secs(5))
+                .expect("every general worker picks up a transfer");
+        }
+
+        // Queue more transfers than there are threads, so the lane is backed up
+        // in the way that used to starve metadata.
+        for _ in 0..16 {
+            let release_rx = release_rx.clone();
+            pool.run(Lane::Transfer, move || {
+                let _ = release_rx.lock().recv();
+            });
+        }
+
+        let (meta_tx, meta_rx) = mpsc::channel();
+        pool.run(Lane::Meta, move || meta_tx.send(()).unwrap());
+        meta_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("a metadata job runs while every transfer thread is stuck");
+
+        for _ in 0..(general + 16) {
+            let _ = release_tx.send(());
+        }
+    }
+
+    /// The reverse direction is allowed and matters for throughput: with no
+    /// transfers in flight, a general worker serves metadata rather than idling.
+    #[test]
+    fn general_workers_fall_back_to_metadata() {
+        // One general worker and no reserved ones, so anything that completes
+        // 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());
+        rx.recv_timeout(Duration::from_secs(5))
+            .expect("a general worker takes metadata when no transfer is waiting");
+    }
+
+    /// A panicking job costs its caller an EIO, never the worker.
+    #[test]
+    fn a_panicking_job_does_not_retire_its_worker() {
+        let pool = Workers::new(1).unwrap();
+        pool.run(Lane::Meta, || panic!("boom"));
+        let (tx, rx) = mpsc::channel();
+        pool.run(Lane::Meta, move || tx.send(()).unwrap());
+        rx.recv_timeout(Duration::from_secs(5))
+            .expect("the pool still serves work after a panic");
+    }
+}

crates/pdfs-gui/src/app/pages/photos.rs

@@ -615,13 +615,6 @@ pub(crate) fn wire_gallery(ui: &Rc<Ui>, list: &gtk4::ListView, scroll: &gtk4::Sc
                     .and_then(|n| n.to_str())
                     .unwrap_or("photo.jpg")
                     .to_string();
-                let bytes = match std::fs::read(&path) {
-                    Ok(b) => b,
-                    Err(e) => {
-                        tracing::error!("Failed to read photo: {e}");
-                        return;
-                    }
-                };
                 let ext = path
                     .extension()
                     .and_then(|e| e.to_str())
@@ -643,10 +636,13 @@ pub(crate) fn wire_gallery(ui: &Rc<Ui>, list: &gtk4::ListView, scroll: &gtk4::Sc
                 ui.busy_begin();
                 let rx = spawn_request(
                     ui.dirs.control_socket(),
+                    // The daemon opens the file itself: it runs on this machine,
+                    // and a photo's bytes through line-delimited JSON is an OOM
+                    // of both processes (see `Request::UploadPhoto`).
                     Request::UploadPhoto {
                         name,
                         media_type: media_type.to_string(),
-                        bytes,
+                        source_path: path.display().to_string(),
                         capture_time,
                     },
                 );

crates/pdfs-gui/src/main.rs

@@ -270,6 +270,12 @@ fn main() -> Result<()> {
     let _ = std::fs::remove_file(&tray_sock);
     let _lock_socket = std::os::unix::net::UnixListener::bind(&tray_sock)
         .context("failed to bind tray single-instance socket")?;
+    // Owner-only like the control socket (B6). This one only guards single
+    // instancing, so a failure is logged rather than fatal — but there is no
+    // reason to leave it reachable by other local users either.
+    if let Err(e) = pdfs_core::config::restrict_socket(&tray_sock) {
+        tracing::warn!(error = %e, "could not restrict tray socket permissions");
+    }
 
     let socket = dirs.control_socket();
     let default_mountpoint = dirs.default_mountpoint();