fix(fs) performance, locks

crates/pdfs-fuse/src/background.rs

@@ -8,35 +8,20 @@ use super::*;
 ///
 /// Returns whether this mount actually held the node, so the caller can evict
 /// its content blob once if *any* mount did.
-fn forget_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) -> bool {
+fn forget_and_notify(st: &mut State, notify: &mut NotifyBatch, uid: &NodeUid) -> bool {
     // Capture the inode before `forget` clears the uid mapping.
     let child = st.by_uid.get(uid).copied();
     let Some((parent, name)) = st.forget(uid) else {
         return false;
     };
-    if let Some(notifier) = notifier {
-        match child {
-            Some(child) => {
-                let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name));
-            }
-            None => {
-                let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name));
-            }
-        }
+    match child {
+        Some(child) => notify.delete(parent, child, name),
+        None => notify.inval_entry(parent, name),
     }
     true
 }
 
-fn invalidate_access_changes(notifier: Option<&Notifier>, changed: &[u64]) {
-    let Some(notifier) = notifier else {
-        return;
-    };
-    for &ino in changed {
-        let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
-    }
-}
-
-fn hide_foreign_deleted_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) {
+fn hide_foreign_deleted_and_notify(st: &mut State, notify: &mut NotifyBatch, uid: &NodeUid) {
     let child = st.by_uid.get(uid).copied();
     let dentry = child.and_then(|ino| {
         st.entries
@@ -45,9 +30,9 @@ fn hide_foreign_deleted_and_notify(st: &mut State, notifier: Option<&Notifier>,
     });
     let changed = st.downgrade_shared_subtree(uid);
     st.hide_foreign_subtree(uid);
-    invalidate_access_changes(notifier, &changed);
-    if let (Some(notifier), Some((parent, name)), Some(child)) = (notifier, dentry, child) {
-        let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name));
+    notify.extend_inodes(&changed);
+    if let (Some((parent, name)), Some(child)) = (dentry, child) {
+        notify.delete(parent, child, name);
     }
 }
 
@@ -132,18 +117,18 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf
         && foreign_delete
     {
         apply_foreign_delete(&core.db, node_uid, |uid| {
-            core.for_each_mount(|st, notifier| {
-                hide_foreign_deleted_and_notify(st, notifier, uid);
+            core.for_each_mount(|st, notify| {
+                hide_foreign_deleted_and_notify(st, notify, uid);
             });
         })
     } else {
         apply_access_downgrade(&core.db, event, |root| {
-            core.for_each_mount(|st, notifier| {
+            core.for_each_mount(|st, notify| {
                 let changed = match root {
                     Some(uid) => st.downgrade_shared_subtree(uid),
                     None => st.downgrade_known_shared_access(),
                 };
-                invalidate_access_changes(notifier, &changed);
+                notify.extend_inodes(&changed);
             });
         })
     };
@@ -173,19 +158,17 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf
                 debug!(uid = %node_uid, "ignoring remote event for a node with a queued write");
             }
             let mut had_node = false;
-            core.for_each_mount(|st, notifier| {
+            core.for_each_mount(|st, notify| {
                 if *is_trashed {
                     // Trashing makes a node vanish from its parent listing.
-                    had_node |= forget_and_notify(st, notifier, node_uid);
+                    had_node |= forget_and_notify(st, notify, node_uid);
                 } else if !ours && let Some(&ino) = st.by_uid.get(node_uid) {
                     // Known node changed: drop its cached attrs/data (and
                     // listing if it is a directory) so the next access
                     // re-fetches. Its content blob may now be stale too.
                     had_node = true;
                     st.invalidate_listing(ino);
-                    if let Some(notifier) = notifier {
-                        let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
-                    }
+                    notify.inval_inode(ino);
                 }
             });
             // Once, not once per mount: the content cache is per-daemon.
@@ -209,8 +192,8 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf
         DriveEvent::NodeDeleted { node_uid, .. } => {
             core.cache.evict(node_uid);
             if !foreign_delete {
-                core.for_each_mount(|st, notifier| {
-                    forget_and_notify(st, notifier, node_uid);
+                core.for_each_mount(|st, notify| {
+                    forget_and_notify(st, notify, node_uid);
                 });
             }
         }
@@ -223,13 +206,11 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf
         | DriveEvent::SharedWithMeUpdated { .. } => {
             warn!("event access continuity lost; downgrading shared trees and resyncing lazily");
             apply_result = apply_result.and(core.db.clear_state(SHARED_WITH_ME_SYNCED_MS));
-            core.for_each_mount(|st, notifier| {
+            core.for_each_mount(|st, notify| {
                 let dirs: Vec<u64> = st.children.keys().copied().collect();
                 for &ino in &dirs {
                     st.invalidate_listing(ino);
-                    if let Some(notifier) = notifier {
-                        let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
-                    }
+                    notify.inval_inode(ino);
                 }
             });
         }
@@ -303,18 +284,16 @@ fn flush_dirty_parents(core: &Core, dirty: &DirtyParents) {
     if dirty.foreign.is_empty() && dirty.ours.is_empty() {
         return;
     }
-    core.for_each_mount(|st, notifier| {
-        let drop_listing = |st: &mut State, parent_uid: &NodeUid| {
-            let Some(&parent) = st.by_uid.get(parent_uid) else {
-                return;
-            };
-            st.invalidate_listing(parent);
-            if let Some(notifier) = notifier {
-                let _ = notifier.inval_inode(INodeNo(parent), 0, 0);
-            }
+    fn drop_listing(st: &mut State, notify: &mut NotifyBatch, parent_uid: &NodeUid) {
+        let Some(&parent) = st.by_uid.get(parent_uid) else {
+            return;
         };
+        st.invalidate_listing(parent);
+        notify.inval_inode(parent);
+    }
+    core.for_each_mount(|st, notify| {
         for parent_uid in &dirty.foreign {
-            drop_listing(st, parent_uid);
+            drop_listing(st, notify, parent_uid);
         }
         for (parent_uid, children) in &dirty.ours {
             if dirty.foreign.contains(parent_uid) {
@@ -328,7 +307,7 @@ fn flush_dirty_parents(core: &Core, dirty: &DirtyParents) {
             {
                 continue;
             }
-            drop_listing(st, parent_uid);
+            drop_listing(st, notify, parent_uid);
         }
     });
 }

crates/pdfs-fuse/src/control.rs

@@ -29,7 +29,23 @@ use super::{Core, count_noun, human_bytes, human_duration, parse_uid};
 
 const MAX_CONTROL_REQUEST_BYTES: u64 = 1024 * 1024;
 const CONTROL_READ_TIMEOUT: Duration = Duration::from_secs(10);
+/// A client that connects, sends a request and then never reads its reply used
+/// to park a handler thread forever in `write_all`; 64 of those exhaust
+/// [`MAX_CONTROL_HANDLERS`] permanently. Generous enough for the largest
+/// responses (photo bytes) over a local socket, finite so a wedged peer can't
+/// hold a permit.
+const CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(30);
 const MAX_CONTROL_HANDLERS: usize = 64;
+/// Ceiling on any client-supplied row `limit`. Every in-tree caller asks for far
+/// less (200 for activity, 60 for a photo page), but the value goes straight
+/// into `SQL LIMIT ?` under the daemon's single connection — an unclamped one
+/// materialises the whole account, and then serialises it into one JSON line.
+const MAX_ROW_LIMIT: usize = 1000;
+
+/// Clamp a client-supplied row limit to [`MAX_ROW_LIMIT`].
+fn clamp_limit(limit: usize) -> usize {
+    limit.min(MAX_ROW_LIMIT)
+}
 static ACTIVE_CONTROL_HANDLERS: AtomicUsize = AtomicUsize::new(0);
 
 struct ControlHandlerPermit;
@@ -109,6 +125,10 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
         warn!(error = %e, "control: could not set request timeout");
         return;
     }
+    if let Err(e) = stream.set_write_timeout(Some(CONTROL_WRITE_TIMEOUT)) {
+        warn!(error = %e, "control: could not set response timeout");
+        return;
+    }
     let mut reader = BufReader::new(match stream.try_clone() {
         Ok(s) => s,
         Err(e) => {
@@ -204,7 +224,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             kind,
             range,
             favorites,
-        }) => match core.photos_timeline(offset, limit, kind, range, favorites) {
+        }) => match core.photos_timeline(offset, clamp_limit(limit), kind, range, favorites) {
             Ok(Some(items)) => CtlResponse::Photos {
                 available: true,
                 items,
@@ -245,7 +265,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             // An album page is a timeline page as far as the front-end is
             // concerned, so it comes back in the same reply — with `counts`
             // unset, which describe the whole timeline rather than this album.
-            Some(album) => match core.album_photos(&album, offset, limit) {
+            Some(album) => match core.album_photos(&album, offset, clamp_limit(limit)) {
                 Ok(items) => CtlResponse::Photos {
                     available: true,
                     items,
@@ -346,22 +366,24 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
                 Err(e) => CtlResponse::error(e),
             }
         }
-        Ok(CtlRequest::Search { query, limit }) => match core.search(&query, limit) {
+        Ok(CtlRequest::Search { query, limit }) => match core.search(&query, clamp_limit(limit)) {
             Ok(hits) => CtlResponse::SearchResults { hits },
             Err(e) => CtlResponse::error(e),
         },
-        Ok(CtlRequest::SearchLocal { query, limit }) => match core.search_local(&query, limit) {
-            Ok(hits) => CtlResponse::LocalResults {
-                hits,
-                indexing: core.indexing.load(Ordering::Relaxed),
-            },
-            Err(e) => CtlResponse::error(e),
-        },
+        Ok(CtlRequest::SearchLocal { query, limit }) => {
+            match core.search_local(&query, clamp_limit(limit)) {
+                Ok(hits) => CtlResponse::LocalResults {
+                    hits,
+                    indexing: core.indexing.load(Ordering::Relaxed),
+                },
+                Err(e) => CtlResponse::error(e),
+            }
+        }
         Ok(CtlRequest::SearchV2 {
             query,
             limit,
             filters,
-        }) => match core.search_v2(&query, limit, &filters) {
+        }) => match core.search_v2(&query, clamp_limit(limit), &filters) {
             Ok((drive_hits, local_hits)) => CtlResponse::SearchResultsV2 {
                 drive_hits,
                 local_hits,
@@ -1023,7 +1045,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             Err(e) => CtlResponse::error(e),
         },
         Ok(CtlRequest::ListActivity { limit }) => CtlResponse::Activity {
-            items: core.list_activity(limit),
+            items: core.list_activity(clamp_limit(limit)),
         },
         // The request did not parse: the caller sent something malformed.
         Err(e) => CtlResponse::error(CoreError::invalid(format!("bad request: {e}"))),

crates/pdfs-fuse/src/devices.rs

@@ -536,7 +536,14 @@ impl Core {
         // folder the daemon no longer tracks, and nothing would ever unmount it.
         // Tear it down first — including before trashing the remote tree it
         // serves, which would otherwise leave it answering for deleted nodes.
-        if let Some(mount) = self.mounts.lock().remove(&id)
+        // Taken in its own statement so the registry lock is released before the
+        // teardown: an edition-2024 `if let` chain holds the guard for the whole
+        // body, and `teardown` unmounts and joins the session thread — which can
+        // be inside a multi-second transfer. The shutdown path takes the same
+        // lock, so holding it here risks systemd's stop timeout landing mid-
+        // unmount and stranding the endpoint.
+        let taken = self.mounts.lock().remove(&id);
+        if let Some(mount) = taken
             && let Err(error) = commit_after_teardown(mount.teardown(), || ())
         {
             let _ = self.db.sync_folder_set_state(id, "error", now_secs());
@@ -866,8 +873,11 @@ impl Core {
                 Ok(format!("{} is now on-demand", local.display()))
             }
             ApplicableMountMode::Mirror => {
-                // ondemand→mirror: tear down the secondary mount first.
-                if let Some(mount) = self.mounts.lock().remove(&id)
+                // ondemand→mirror: tear down the secondary mount first. Taken
+                // before the teardown so the registry lock is not held across
+                // the unmount and thread join (see `remove_sync_folder`).
+                let taken = self.mounts.lock().remove(&id);
+                if let Some(mount) = taken
                     && let Err(error) = commit_after_teardown(mount.teardown(), || ())
                 {
                     let _ = self.db.sync_folder_set_state(id, "error", now_secs());

crates/pdfs-fuse/src/drain.rs

@@ -136,6 +136,10 @@ impl Core {
             let due = self.db.next_due_op(now).unwrap_or_default();
 
             let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else {
+                // Nothing to upload is exactly when the connection is free, so
+                // this is where the read path's buffered LRU touches get written
+                // (`ContentCache::flush_touches`). Cheap and a no-op when empty.
+                self.cache.flush_touches();
                 self.recover_fsynced_writes();
                 // A debounced or backed-off op may be waiting: sleep only
                 // until it becomes due rather than the full idle-poll.
@@ -709,8 +713,9 @@ impl Core {
         self.db.delete_op(op.id)?;
         // Dropping the pending entry hands the node back to the remote's truth:
         // reads stop coming from the staged blob, and the event sync stops
-        // skipping it as "ahead of the server" (offline.md Phase 3a).
-        self.pending.lock().remove(uid);
+        // skipping it as "ahead of the server" (offline.md Phase 3a). Only if it
+        // is still *this* write's entry — see [`Core::release_pending`].
+        self.release_pending(uid, blob);
         self.cache.discard_staged(blob);
         self.cache.evict(uid);
         self.evict_reader(uid);
@@ -748,7 +753,7 @@ impl Core {
         error!(%uid, name, reason, staged = %blob.display(),
                "cannot place a conflicted write; bytes kept in staging");
         self.db.delete_op(op.id)?;
-        self.pending.lock().remove(uid);
+        self.release_pending(uid, blob);
         self.cache.evict(uid);
         self.evict_reader(uid);
         self.log_activity(
@@ -760,6 +765,24 @@ impl Core {
         Ok(())
     }
 
+    /// Drop `uid`'s pending entry, but only while it still names `blob`.
+    ///
+    /// A supersede can replace the entry while the op that owned it is in
+    /// flight, so an unconditional `remove` retires a *newer* write's staged
+    /// blob: reads fall back to a stale remote, the next `remote_baseline` forks
+    /// a spurious `(sync-conflict)` copy of bytes that were never in conflict,
+    /// and the incomplete-write interlock — which keys off the pending entry —
+    /// is disarmed, letting a gap-fill read from a revision that no longer
+    /// describes the file.
+    pub(crate) fn release_pending(&self, uid: &NodeUid, blob: &Path) {
+        let mut pending = self.pending.lock();
+        if let Some(current) = pending.get(uid)
+            && current.path == blob
+        {
+            pending.remove(uid);
+        }
+    }
+
     /// Whether a folder a queued op targets has stopped being a place a node can
     /// go. A failure to ask is not an answer: only a definite "trashed" or "not
     /// there" counts, so a network fault leaves the op to retry normally.
@@ -886,14 +909,7 @@ impl Core {
         // an orphaned file (harmless), whereas the reverse would leave a queued
         // op pointing at nothing.
         self.db.delete_op(op.id)?;
-        {
-            let mut p = self.pending.lock();
-            if let Some(op_in_mem) = p.get(&uid)
-                && op_in_mem.path == blob
-            {
-                p.remove(&uid);
-            }
-        }
+        self.release_pending(&uid, &blob);
 
         // 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.

crates/pdfs-fuse/src/filesystem.rs

@@ -3,6 +3,73 @@
 use super::*;
 
 impl Filesystem for ProtonFs {
+    /// Negotiate kernel-side behaviour once, before any other callback.
+    ///
+    /// Everything here is a hint the kernel is free to decline, so each is
+    /// requested and its refusal logged rather than treated as fatal — an older
+    /// kernel that lacks a capability must still mount.
+    ///
+    /// Deliberately *not* requested: `FUSE_ATOMIC_O_TRUNC` and
+    /// `FUSE_DO_READDIRPLUS` change what the kernel sends (truncation folded
+    /// into `open`, `readdirplus` instead of `readdir` + N lookups), so they
+    /// need the matching handlers first; and `FOPEN_KEEP_CACHE` needs the write
+    /// path to invalidate page cache on every locally staged revision, not just
+    /// on remote events.
+    fn init(&mut self, _req: &Request, config: &mut KernelConfig) -> std::io::Result<()> {
+        // Readahead the kernel issues per file. Matching the block size means a
+        // sequential reader's readahead lands inside the block a demand read is
+        // already fetching, instead of one 128 KiB slice at a time.
+        if let Err(nearest) = config.set_max_readahead(BLOCK_SIZE as u32) {
+            debug!(nearest, "kernel capped max_readahead");
+            let _ = config.set_max_readahead(nearest);
+        }
+        if let Err(nearest) = config.set_max_background(MAX_BACKGROUND) {
+            debug!(nearest, "kernel capped max_background");
+        }
+        // Without this the kernel serialises lookups within one directory, so a
+        // cold `ls` of a folder is a chain of round trips rather than a fan-out.
+        if let Err(unsupported) = config.add_capabilities(InitFlags::FUSE_PARALLEL_DIROPS) {
+            debug!(?unsupported, "kernel does not support parallel dirops");
+        }
+        if let Err(unsupported) = config.add_capabilities(InitFlags::FUSE_CACHE_SYMLINKS) {
+            debug!(?unsupported, "kernel does not support symlink caching");
+        }
+        Ok(())
+    }
+
+    /// Free and total space, as `df` and every "will this fit?" preflight ask.
+    ///
+    /// Reported in [`REPORTED_BLKSIZE`] units against the *account* quota, not
+    /// the local cache: what a copy into the mount consumes is Drive storage.
+    /// The default implementation answers all-zeroes, which reads as a full
+    /// filesystem and makes installers and file managers refuse to write.
+    ///
+    /// On a worker, because a cold answer is a round trip
+    /// ([`Core::account_quota_cached`] serves a warm one from memory).
+    fn statfs(&self, _req: &Request, _ino: INodeNo, reply: ReplyStatfs) {
+        let core = self.core.clone();
+        self.core.workers.run(Lane::Meta, move || {
+            let (max_space, used_space) = core.account_quota_cached().unwrap_or((0, 0));
+            let bsize = u64::from(REPORTED_BLKSIZE);
+            let blocks = (max_space.max(0) as u64).div_ceil(bsize);
+            let used = (used_space.max(0) as u64).div_ceil(bsize);
+            let free = blocks.saturating_sub(used);
+            // `files`/`ffree` are unknowable — Drive has no inode budget — so
+            // they are reported as "plenty" rather than as zero, which some
+            // tools read as "no more files may be created".
+            reply.statfs(
+                blocks,
+                free,
+                free,
+                0,
+                u64::MAX,
+                REPORTED_BLKSIZE,
+                MAX_NAME_LEN,
+                REPORTED_BLKSIZE,
+            );
+        });
+    }
+
     fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
         let parent = parent.0;
         let name = match fuse_name(name) {
@@ -109,109 +176,19 @@ impl Filesystem for ProtonFs {
             .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply));
     }
 
+    /// A read-only open is a map lookup and answers inline; a write open has to
+    /// allocate a scratch file and, when a queued revision is the base, duplicate
+    /// a blob that is the whole file — so it is served from a worker rather than
+    /// stalling every other request on the mount behind it.
     fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
-        let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY;
-        let (uid, base_mtime, base_size, base_revision_id) = {
-            let mut st = self.core.state.lock();
-            match st.entries.get_mut(&ino.0) {
-                Some(e) if e.node.is_file() => {
-                    if write_requested && !e.writable() {
-                        reply.error(Errno::EACCES);
-                        return;
-                    }
-                    e.open_count = e.open_count.saturating_add(1);
-                    (
-                        e.uid.clone(),
-                        e.node.modification_time,
-                        node_size(&e.node),
-                        node_revision_id(&e.node),
-                    )
-                }
-                Some(_) => {
-                    reply.error(Errno::EISDIR);
-                    return;
-                }
-                None => {
-                    reply.error(Errno::ENOENT);
-                    return;
-                }
-            }
-        };
-        debug!(ino = ino.0, ?flags, base_size, base_mtime, "open");
-        // Read-only opens stay stateless (fh 0). A write open allocates a
-        // disk-backed handle; bytes are authored into a scratch file and the
-        // untouched remainder is pulled from the base lazily (on read / commit).
         if flags.acc_mode() == OpenAccMode::O_RDONLY {
-            reply.opened(FileHandle(0), FopenFlags::empty());
+            self.serve_open(ino, flags, reply);
             return;
         }
-        // A previous close may already have newer bytes queued locally while
-        // the server still holds the older revision. That staged blob, not the
-        // remote, is the base for this handle. In particular, `truncate file`
-        // opens for writing and then shrinks via setattr; starting its scratch
-        // as a sparse zero file would turn the preserved prefix into zeros.
-        //
-        // Complete pending blobs are the common case and can be copied without
-        // any network access. An incomplete blob still has gaps referring to
-        // the remote base; stacking another write on it cannot be represented
-        // safely by WriteHandle, so fail the open rather than risk corruption.
-        let pending_base = self.core.pending.lock().get(&uid).cloned();
-        if pending_base.as_ref().is_some_and(|p| !p.meta.complete) {
-            if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
-                entry.open_count = entry.open_count.saturating_sub(1);
-            }
-            error!(%uid, "refusing write over incomplete queued revision");
-            reply.error(Errno::EIO);
-            return;
-        }
-        let (file, path) = match self.core.cache.create_scratch() {
-            Ok(x) => x,
-            Err(e) => {
-                if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
-                    entry.open_count = entry.open_count.saturating_sub(1);
-                }
-                error!(%uid, error = %e, "create scratch file failed");
-                reply.error(Errno::EIO);
-                return;
-            }
-        };
-        let mut initial_written = Intervals::default();
-        if let Some(pending) = &pending_base {
-            if let Err(e) = std::fs::copy(&pending.path, &path) {
-                if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
-                    entry.open_count = entry.open_count.saturating_sub(1);
-                }
-                let _ = std::fs::remove_file(&path);
-                error!(%uid, source = %pending.path.display(), error = %e,
-                    "copy queued revision into write scratch failed");
-                reply.error(Errno::EIO);
-                return;
-            }
-            initial_written.add(0, pending.meta.len);
-        }
-        let mut st = self.core.state.lock();
-        let fh = st.next_fh;
-        st.next_fh += 1;
-        let aw = st.active_writes.entry(ino.0).or_insert_with(|| {
-            WriteHandle {
-                ino: ino.0,
-                uid,
-                file: Arc::new(file),
-                path,
-                written: initial_written,
-                // Starts at the current size; reads in [0, base_size) come from
-                // the base until overwritten.
-                len: base_size,
-                base_size,
-                base_mtime,
-                base_revision_id,
-                dirty: false,
-                open_count: 0,
-            }
-        });
-        aw.open_count += 1;
-        st.handles.insert(fh, ino.0);
-        reply.opened(FileHandle(fh), FopenFlags::empty());
+        let fs = self.clone();
+        self.core
+            .workers
+            .run(Lane::Transfer, move || fs.serve_open(ino, flags, reply));
     }
 
     fn read(
@@ -374,7 +351,16 @@ impl Filesystem for ProtonFs {
         };
         if let Err(e) = file.write_all_at(data, offset) {
             error!(ino = ino.0, fh, error = %e, "scratch write failed");
-            reply.error(Errno::EIO);
+            // A full local disk is not a broken file: answer `ENOSPC` so the
+            // application can say so (and so `cp`/editors take their own
+            // out-of-space path), and free what the cache is holding — the
+            // scratch file the next write goes to shares that filesystem.
+            if e.raw_os_error() == Some(libc::ENOSPC) {
+                self.core.cache.emergency_evict();
+                reply.error(Errno::ENOSPC);
+            } else {
+                reply.error(Errno::EIO);
+            }
             return;
         }
         let new_len = {
@@ -566,6 +552,12 @@ impl Filesystem for ProtonFs {
     /// for fsync wants its bytes to survive a crash, and blocking it on an upload
     /// (which offline would never finish) buys nothing the queue does not already
     /// guarantee.
+    ///
+    /// Served on a worker: `sync_all` is a disk barrier and the sidecar write is
+    /// two more file operations, and an application that fsyncs per record would
+    /// otherwise stall every request on the mount at its barrier rate. Nothing
+    /// orders this against other callbacks — the kernel waits for the reply
+    /// either way, which is the whole contract.
     fn fsync(
         &self,
         _req: &Request,
@@ -574,76 +566,10 @@ impl Filesystem for ProtonFs {
         _datasync: bool,
         reply: ReplyEmpty,
     ) {
-        // Everything the durability sidecar needs, copied out so no lock is held
-        // across the sync.
-        let handle = {
-            let st = self.core.state.lock();
-            let ino = st.handles.get(&fh.0).copied();
-            ino.and_then(|i| st.active_writes.get(&i)).map(|aw| {
-                (
-                    aw.file.clone(),
-                    aw.path.clone(),
-                    aw.uid.clone(),
-                    aw.dirty,
-                    aw.written.clone(),
-                    aw.len,
-                    aw.base_size,
-                    aw.base_mtime,
-                    aw.base_revision_id.clone(),
-                )
-            })
-        };
-        let Some((f, path, uid, dirty, written, len, base_size, base_mtime, base_revision_id)) =
-            handle
-        else {
-            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, base_revision_id),
-        };
-        // 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();
+        let fs = self.clone();
+        self.core
+            .workers
+            .run(Lane::Transfer, move || fs.serve_fsync(fh, reply));
     }
 
     fn release(
@@ -1146,7 +1072,7 @@ impl ProtonFs {
             uid: self.uid,
             gid: self.gid,
             rdev: 0,
-            blksize: 512,
+            blksize: REPORTED_BLKSIZE,
             flags: 0,
         }
     }
@@ -1154,6 +1080,202 @@ impl ProtonFs {
     /// Trash the child `name` under `parent` on the remote, then drop it from the
     /// local cache. Backs both `unlink` and `rmdir` (Proton trashes whole
     /// subtrees, so an `rmdir` of a non-empty dir behaves the same).
+    /// The body of [`Filesystem::open`], run off the dispatch loop for a write
+    /// open.
+    fn serve_open(&self, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
+        let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY;
+        let (uid, base_mtime, base_size, base_revision_id) = {
+            let mut st = self.core.state.lock();
+            match st.entries.get_mut(&ino.0) {
+                Some(e) if e.node.is_file() => {
+                    if write_requested && !e.writable() {
+                        reply.error(Errno::EACCES);
+                        return;
+                    }
+                    e.open_count = e.open_count.saturating_add(1);
+                    (
+                        e.uid.clone(),
+                        e.node.modification_time,
+                        node_size(&e.node),
+                        node_revision_id(&e.node),
+                    )
+                }
+                Some(_) => {
+                    reply.error(Errno::EISDIR);
+                    return;
+                }
+                None => {
+                    reply.error(Errno::ENOENT);
+                    return;
+                }
+            }
+        };
+        debug!(ino = ino.0, ?flags, base_size, base_mtime, "open");
+        // Read-only opens stay stateless (fh 0). A write open allocates a
+        // disk-backed handle; bytes are authored into a scratch file and the
+        // untouched remainder is pulled from the base lazily (on read / commit).
+        if flags.acc_mode() == OpenAccMode::O_RDONLY {
+            reply.opened(FileHandle(0), FopenFlags::empty());
+            return;
+        }
+        // A previous close may already have newer bytes queued locally while
+        // the server still holds the older revision. That staged blob, not the
+        // remote, is the base for this handle. In particular, `truncate file`
+        // opens for writing and then shrinks via setattr; starting its scratch
+        // as a sparse zero file would turn the preserved prefix into zeros.
+        //
+        // Complete pending blobs are the common case and can be copied without
+        // any network access. An incomplete blob still has gaps referring to
+        // the remote base; stacking another write on it cannot be represented
+        // safely by WriteHandle, so fail the open rather than risk corruption.
+        let pending_base = self.core.pending.lock().get(&uid).cloned();
+        if pending_base.as_ref().is_some_and(|p| !p.meta.complete) {
+            if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
+                entry.open_count = entry.open_count.saturating_sub(1);
+            }
+            error!(%uid, "refusing write over incomplete queued revision");
+            reply.error(Errno::EIO);
+            return;
+        }
+        let (file, path) = match self.core.cache.create_scratch() {
+            Ok(x) => x,
+            Err(e) => {
+                if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
+                    entry.open_count = entry.open_count.saturating_sub(1);
+                }
+                error!(%uid, error = %e, "create scratch file failed");
+                reply.error(Errno::EIO);
+                return;
+            }
+        };
+        let mut initial_written = Intervals::default();
+        if let Some(pending) = &pending_base {
+            // Reflink where the filesystem can: this blob is the whole file, and
+            // a byte-for-byte copy of a multi-GB pending revision is the kind of
+            // work `open` must not do (`ContentCache::clone_or_copy`).
+            if let Err(e) = ContentCache::clone_or_copy(&pending.path, &path) {
+                if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
+                    entry.open_count = entry.open_count.saturating_sub(1);
+                }
+                let _ = std::fs::remove_file(&path);
+                error!(%uid, source = %pending.path.display(), error = %e,
+                    "copy queued revision into write scratch failed");
+                reply.error(Errno::EIO);
+                return;
+            }
+            initial_written.add(0, pending.meta.len);
+        }
+        let mut st = self.core.state.lock();
+        let fh = st.next_fh;
+        st.next_fh += 1;
+        let aw = st.active_writes.entry(ino.0).or_insert_with(|| {
+            WriteHandle {
+                ino: ino.0,
+                uid,
+                file: Arc::new(file),
+                path,
+                written: initial_written,
+                // Starts at the current size; reads in [0, base_size) come from
+                // the base until overwritten.
+                len: base_size,
+                base_size,
+                base_mtime,
+                base_revision_id,
+                dirty: false,
+                open_count: 0,
+            }
+        });
+        aw.open_count += 1;
+        st.handles.insert(fh, ino.0);
+        reply.opened(FileHandle(fh), FopenFlags::empty());
+    }
+
+    /// The body of [`Filesystem::fsync`], run off the dispatch loop.
+    fn serve_fsync(&self, fh: FileHandle, reply: ReplyEmpty) {
+        // Everything the durability sidecar needs, copied out so no lock is held
+        // across the sync.
+        let handle = {
+            let st = self.core.state.lock();
+            let ino = st.handles.get(&fh.0).copied();
+            ino.and_then(|i| st.active_writes.get(&i)).map(|aw| {
+                (
+                    aw.file.clone(),
+                    aw.path.clone(),
+                    aw.uid.clone(),
+                    aw.dirty,
+                    aw.written.clone(),
+                    aw.len,
+                    aw.base_size,
+                    aw.base_mtime,
+                    aw.base_revision_id.clone(),
+                )
+            })
+        };
+        let Some((f, path, uid, dirty, written, len, base_size, base_mtime, base_revision_id)) =
+            handle
+        else {
+            reply.error(Errno::EBADF);
+            return;
+        };
+        if let Err(e) = f.sync_all() {
+            error!(fh = fh.0, error = %e, "fsync of scratch file failed");
+            // Delayed allocation means a full disk usually surfaces here rather
+            // than at `write(2)`; `ENOSPC` is what an application checks for.
+            if e.raw_os_error() == Some(libc::ENOSPC) {
+                self.core.cache.emergency_evict();
+                reply.error(Errno::ENOSPC);
+            } else {
+                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, base_revision_id),
+        };
+        // 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");
+            if e.is_disk_full() {
+                self.core.cache.emergency_evict();
+                reply.error(Errno::ENOSPC);
+            } else {
+                reply.error(Errno::EIO);
+            }
+            return;
+        }
+        reply.ok();
+    }
+
     /// The body of [`Filesystem::create`], run off the dispatch loop.
     fn serve_create(&self, parent: u64, name: &str, reply: ReplyCreate) {
         if let Err(error) = self.core.require_writable(parent) {

crates/pdfs-fuse/src/lib.rs

@@ -48,10 +48,10 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use fuser::ReplyXattr;
 use fuser::{
     AccessFlags, BackgroundSession, BsdFileFlags, Config, CopyFileRangeFlags, Errno, FileAttr,
-    FileHandle, FileType, Filesystem, FopenFlags, Generation, INodeNo, IoctlFlags, LockOwner,
-    MountOption, Notifier, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr, ReplyCreate, ReplyData,
-    ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyLseek, ReplyOpen, ReplyWrite, Request,
-    Session, TimeOrNow, WriteFlags,
+    FileHandle, FileType, Filesystem, FopenFlags, Generation, INodeNo, InitFlags, IoctlFlags,
+    KernelConfig, LockOwner, MountOption, Notifier, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr,
+    ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyLseek,
+    ReplyOpen, ReplyStatfs, ReplyWrite, Request, Session, TimeOrNow, WriteFlags,
 };
 use futures::StreamExt as _;
 use pdfs_core::batch;
@@ -121,6 +121,27 @@ use workers::{FUSE_WORKERS, Lane, Workers};
 /// change this is how long the kernel may serve stale metadata.
 const TTL: Duration = Duration::from_secs(30);
 
+/// How long a `statfs(2)` answer is served from [`Core::quota`] before the
+/// account figures are refetched. See [`Core::account_quota_cached`].
+const QUOTA_TTL: Duration = Duration::from_secs(60);
+
+/// Block size reported to the kernel and to `statfs(2)`.
+///
+/// coreutils and most copy tools size their I/O buffers from `st_blksize`, and
+/// the fuser default of 512 makes them issue reads two orders of magnitude
+/// smaller than the [`BLOCK_SIZE`] this filesystem actually fetches in. 1 MiB is
+/// the largest value that stays well inside the kernel's per-request limit.
+const REPORTED_BLKSIZE: u32 = 1 << 20;
+
+/// Ceiling on in-flight background requests (readahead, writeback). The kernel
+/// default of 16 caps how much of a sequential read the kernel will run ahead of
+/// the application, which on a high-latency link is the whole ballgame.
+const MAX_BACKGROUND: u16 = 64;
+
+/// Longest single path component, as reported by `statfs(2)`. The kernel's own
+/// dirent limit; Proton's is not lower.
+const MAX_NAME_LEN: u32 = 255;
+
 /// How often the background task polls the remote event cursor.
 const POLL_INTERVAL: Duration = Duration::from_secs(10);
 /// First and longest delay between probes for the network coming back after an
@@ -396,6 +417,14 @@ struct Core {
     /// is the validity tag — a new revision may well have a thumbnail — matching
     /// how [`ContentCache::read_thumbnail`] validates the positive side.
     no_thumbnail: Arc<Mutex<HashMap<(NodeUid, i32), i64>>>,
+    /// Last account quota answer and when it was learned, for `statfs(2)`.
+    ///
+    /// `df` and the free-space preflight of every file manager and installer ask
+    /// for this, and the answer costs a round trip — so it is served from here
+    /// for [`QUOTA_TTL`] and refreshed on a worker afterwards. Total account
+    /// storage moves slowly enough that a minute-old number is a better answer
+    /// than either a stall or the zeroes the default implementation returns.
+    quota: Arc<Mutex<Option<(std::time::Instant, i64, i64)>>>,
     /// Folders whose listing was enumerated cheaply and is having its file sizes
     /// filled in right now, so a burst of `stat`s over a fresh listing starts one
     /// upgrade rather than one per entry. See [`Core::spawn_size_upgrade`].
@@ -701,11 +730,18 @@ impl Core {
     /// notification channel, so work that invalidates cached metadata reaches
     /// the session that actually minted the inodes it is naming.
     ///
-    /// `None` for a mount whose session has not been spawned yet — there is
-    /// nothing to invalidate in a kernel that has never been told about it.
-    fn for_each_mount(&self, mut apply: impl FnMut(&mut State, Option<&Notifier>)) {
+    /// The closure records what the kernel needs told into a [`NotifyBatch`]
+    /// rather than sending it: notifications are flushed after this mount's
+    /// State lock is released, the same rule [`Core::for_each_state`] and
+    /// [`Core::flush_access_changes`] follow.
+    fn for_each_mount(&self, mut apply: impl FnMut(&mut State, &mut NotifyBatch)) {
         for (_, state, notifier, _) in self.states.live() {
-            apply(&mut state.lock(), notifier.get());
+            let mut batch = NotifyBatch::default();
+            {
+                let mut state = state.lock();
+                apply(&mut state, &mut batch);
+            }
+            batch.flush(notifier.get());
         }
     }
 
@@ -763,6 +799,75 @@ fn require_uid_access(db: &Db, uid: &NodeUid, live_access: &[Access]) -> Result<
         .ok_or(Errno::EACCES)
 }
 
+/// One kernel notification, recorded while the State lock is held so it can be
+/// sent once it is not.
+///
+/// `inval_inode` calls into `invalidate_inode_pages2`, which can block waiting
+/// on pages that this daemon's own workers have to fill — and those workers take
+/// the State lock. Sending from inside the lock is that deadlock's shape; a
+/// batch is the smallest way to keep the inode/mount pairing while moving the
+/// send outside it.
+enum KernelNotice {
+    InvalInode(u64),
+    InvalEntry {
+        parent: u64,
+        name: String,
+    },
+    Delete {
+        parent: u64,
+        child: u64,
+        name: String,
+    },
+}
+
+/// Notifications accumulated under one mount's State lock. Flushed by
+/// [`Core::for_each_mount`] against that mount's own channel — inode numbers are
+/// per-mount, so the pairing has to survive the deferral.
+#[derive(Default)]
+struct NotifyBatch(Vec<KernelNotice>);
+
+impl NotifyBatch {
+    fn inval_inode(&mut self, ino: u64) {
+        self.0.push(KernelNotice::InvalInode(ino));
+    }
+
+    fn inval_entry(&mut self, parent: u64, name: String) {
+        self.0.push(KernelNotice::InvalEntry { parent, name });
+    }
+
+    fn delete(&mut self, parent: u64, child: u64, name: String) {
+        self.0.push(KernelNotice::Delete {
+            parent,
+            child,
+            name,
+        });
+    }
+
+    fn extend_inodes(&mut self, inodes: &[u64]) {
+        self.0
+            .extend(inodes.iter().map(|&ino| KernelNotice::InvalInode(ino)));
+    }
+
+    /// Send everything recorded. A mount whose session has not been spawned has
+    /// nothing to tell the kernel, so the batch is simply dropped.
+    fn flush(self, notifier: Option<&Notifier>) {
+        let Some(notifier) = notifier else { return };
+        for notice in self.0 {
+            let _ = match notice {
+                KernelNotice::InvalInode(ino) => notifier.inval_inode(INodeNo(ino), 0, 0),
+                KernelNotice::InvalEntry { parent, name } => {
+                    notifier.inval_entry(INodeNo(parent), OsStr::new(&name))
+                }
+                KernelNotice::Delete {
+                    parent,
+                    child,
+                    name,
+                } => notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)),
+            };
+        }
+    }
+}
+
 fn notify_access_changes(notifier: Option<&Notifier>, changed: &[u64]) {
     let Some(notifier) = notifier else { return };
     for &ino in changed {
@@ -2256,7 +2361,23 @@ impl Core {
         }
         let path = self.cache.stage_write(&meta, src).map_err(|e| {
             error!(%uid, error = %e, "staging write failed");
-            Errno::EIO
+            // The scratch file is still the only copy of these bytes, and the
+            // caller is releasing the handle it belongs to — without a
+            // durability sidecar `discard_unmarked_scratch` deletes it at the
+            // next mount, the one place the "never delete on a failure path"
+            // invariant would break. Mark it first, whatever went wrong.
+            if let Err(mark) = self.cache.mark_scratch_durable(src, &meta) {
+                error!(%uid, error = %mark,
+                       "could not mark an unstageable write durable; bytes at risk");
+            }
+            if e.is_disk_full() {
+                // Free cached blobs so the retry has somewhere to land, and tell
+                // the caller what is actually wrong.
+                self.cache.emergency_evict();
+                Errno::ENOSPC
+            } else {
+                Errno::EIO
+            }
         })?;
         let meta_json = serde_json::to_string(&meta).unwrap_or_default();
         let superseded = if is_local_uid(uid) {
@@ -2310,15 +2431,29 @@ impl Core {
         // Reflect the write in the tree straight away: `ls` must show the new
         // size and mtime even though the remote still holds the old revision.
         let now = now_secs();
-        {
+        let recorded = {
             let mut st = self.state.lock();
-            st.record_pending_write(ino, len, now);
-        }
+            // The only record that the file is as long as the caller was told.
+            // If it cannot be written the write did not fully take, and saying
+            // so beats acknowledging a `close(2)` and then serving the old size
+            // — the bytes themselves are already staged and queued either way.
+            st.record_pending_write(ino, len, now)
+        };
         // Cached blobs and open readers describe the superseded revision. Reads
         // come from the staged blob until the op drains, so just drop them.
+        // Done even when the row above failed: the staged blob is the file's
+        // content now either way, and leaving a stale one cached would serve it.
         self.cache.evict(uid);
         self.evict_reader(uid);
         self.wake_drain();
+        if let Err(e) = recorded {
+            error!(%uid, error = %e, "recording a queued write's size failed");
+            return Err(if e.is_disk_full() {
+                Errno::ENOSPC
+            } else {
+                Errno::EIO
+            });
+        }
         Ok(())
     }
 

crates/pdfs-fuse/src/mount.rs

@@ -255,6 +255,14 @@ pub(super) fn spawn_session(
     root: Node,
 ) -> std::io::Result<BackgroundSession> {
     let mut config = Config::default();
+    // `n_threads` stays at fuser's default of 1 (with `clone_fd` off) on
+    // purpose, for now. A concurrent dispatch loop is worth having, but
+    // `release` currently relies on being serialised against the `open`+`read`
+    // an application issues right after `close(2)` — see the comment on
+    // `Filesystem::release`. Raising it before that ordering is made explicit
+    // (a per-node staging-in-flight barrier) reintroduces the acceptance
+    // suite's "concurrent file 1 mismatch". Kernel-side caps that do *not*
+    // depend on that ordering are negotiated in `ProtonFs::init`.
     config.mount_options = vec![
         MountOption::FSName("protondrive".to_string()),
         MountOption::Subtype("protondrive".to_string()),
@@ -385,6 +393,7 @@ pub fn mount(
         self_changes: Arc::new(Mutex::new(HashMap::new())),
         thumb_gen: Arc::new(Mutex::new(HashSet::new())),
         no_thumbnail: Arc::new(Mutex::new(HashMap::new())),
+        quota: Arc::new(Mutex::new(None)),
         size_upgrades: Arc::new(Mutex::new(HashMap::new())),
         notifier: Arc::new(OnceLock::new()),
         session_live: Arc::new(AtomicBool::new(false)),

crates/pdfs-fuse/src/sharing.rs

@@ -12,6 +12,9 @@ use pdfs_core::control::{
     BookmarkInfo, DirEntry, InvitationInfo, PublicLinkInfo, ShareEntry, ShareEntryKind, SharedItem,
 };
 use pdfs_core::{CoreError, CoreResult};
+use tracing::debug;
+
+use crate::QUOTA_TTL;
 use proton_drive_rs::proton_sdk::ids::NodeUid;
 use proton_drive_rs::{MemberRole, NodeKind};
 
@@ -533,9 +536,33 @@ impl Core {
             .rt
             .block_on(self.client.quota())
             .map_err(|e| CoreError::from_api(&e, "account quota"))?;
+        *self.quota.lock() = Some((std::time::Instant::now(), q.max_space, q.used_space));
         Ok((q.max_space, q.used_space))
     }
 
+    /// The account quota as `statfs(2)` needs it: cached for [`QUOTA_TTL`],
+    /// refetched after that, and `None` when there is no usable answer at all —
+    /// offline with nothing cached.
+    ///
+    /// Never blocks on the network for a *fresh* value when a recent one exists,
+    /// because the callers are `df` and every "is there room?" preflight, and a
+    /// stale byte count is worth more to them than a stalled syscall. Offline,
+    /// the last known figures keep being served rather than degrading to zero.
+    pub(crate) fn account_quota_cached(&self) -> Option<(i64, i64)> {
+        if let Some((at, max_space, used_space)) = *self.quota.lock()
+            && at.elapsed() < QUOTA_TTL
+        {
+            return Some((max_space, used_space));
+        }
+        match self.account_quota() {
+            Ok(q) => Some(q),
+            Err(error) => {
+                debug!(%error, "account quota refresh failed; serving the last known figures");
+                self.quota.lock().map(|(_, max, used)| (max, used))
+            }
+        }
+    }
+
     // ---- shared by me -----------------------------------------------------
 
     /// List the nodes I have shared with others, each with a summary of its share

crates/pdfs-fuse/src/state.rs

@@ -920,14 +920,24 @@ impl State {
     /// file is as long as the caller was told it is. Without it a restart serves
     /// the stale size and the file reads as truncated — or empty — while its
     /// bytes sit safely in staging (offline.md Phase 3).
-    pub(crate) fn record_pending_write(&mut self, ino: u64, size: u64, mtime: i64) {
+    /// Failure is reported rather than logged: dropping the row silently means
+    /// the daemon acknowledged a write and then serves the old size for it, so
+    /// the caller has to be able to turn it into a failed `release`.
+    pub(crate) fn record_pending_write(
+        &mut self,
+        ino: u64,
+        size: u64,
+        mtime: i64,
+    ) -> pdfs_core::Result<()> {
         self.set_size(ino, size);
         self.touch_mtime(ino, mtime);
-        if let Some(e) = self.entries.get(&ino)
-            && let Err(err) = self.db.upsert_node(&e.node)
-        {
-            warn!(uid = %e.uid, error = %err, "db upsert_node failed for a queued write");
+        if let Some(e) = self.entries.get(&ino) {
+            self.db.upsert_node(&e.node).map_err(|err| {
+                warn!(uid = %e.uid, error = %err, "db upsert_node failed for a queued write");
+                err
+            })?;
         }
+        Ok(())
     }
 }
 

crates/pdfs-fuse/src/sweep.rs

@@ -138,9 +138,9 @@ impl Core {
     /// enumerated every node up front and has been doing network I/O since. Before
     /// touching anything remote this re-reads the node and refuses to act unless
     /// it is *still* the same revision, still untrashed, still unwritten and still
-    /// unopened. Without that, a user edit landing inside the window is destroyed
-    /// — `discard_queued_ops` below drops the queued write *and its staged blob*,
-    /// which may hold the only copy of those bytes (`docs/BUGS.md` B71).
+    /// unopened (`docs/BUGS.md` B71). Nothing on this path discards a queued write
+    /// or its staged blob: a write that arrives after that check is left for the
+    /// drain to resolve against the now-trashed node.
     fn remove_conflict_copy(&self, snapshot: &proton_drive_rs::Node, base: &str) {
         let uid = &snapshot.uid;
         let name = snapshot.name.as_str();
@@ -166,10 +166,13 @@ impl Core {
         self.for_each_state(|st| {
             st.forget_or_unlink(uid);
         });
-        if let Err(error) = self.discard_queued_ops(uid) {
-            warn!(%uid, ?error, "conflict sweep: queued-op cleanup failed");
-            return;
-        }
+        // Deliberately *not* `discard_queued_ops`: `duplicate_still_removable`
+        // proved the queue empty before the trash above, but that proof is stale
+        // the moment `trash_nodes` goes to the network. A `close(2)` landing in
+        // that window stages bytes, and discarding here would unlink the staged
+        // blob holding the only copy of them. Leaving the op queued is the
+        // non-destructive outcome: the drain re-reads the node, sees it trashed,
+        // and `revision_conflict` routes it to `keep_as_conflict_copy`.
         self.cache.evict(uid);
         self.evict_reader(uid);
         if let Err(e) = self.db.delete_node(uid) {
@@ -194,8 +197,8 @@ impl Core {
         let uid = &snapshot.uid;
         let name = snapshot.name.as_str();
 
-        // A queued op means bytes are still owed an upload. `discard_queued_ops`
-        // would throw those away along with the staged blob holding them.
+        // A queued op means bytes are still owed an upload; trashing the node
+        // under them turns an ordinary upload into a conflict fork.
         match self.db.has_any_op(&uid.to_string()) {
             Ok(false) => {}
             Ok(true) => {