cli: Fix FUSE kernel cache serving stale directory listings (#317)

penberg · web-flow · commit 4690a34e28d2 · 2026-02-13T10:13:32.000+02:00
When readdirplus populates the kernel dcache with TTL=MAX, the
kernel caches entries indefinitely. After unlink/rmdir/rename the
parent directory's cached readdirplus result can still contain
stale entries on subsequent ls calls.

Fix by sending FUSE_NOTIFY_INVAL_ENTRY after unlink, rmdir, and
rename. Notifications are dispatched from a background thread to
avoid deadlocking the single-threaded session loop.

Add integration test covering unlink, rmdir, rename, create, and
mkdir visibility in subsequent readdir.
diff --git a/cli/src/fuse.rs b/cli/src/fuse.rs
@@ -68,7 +68,8 @@ fn maximize_fd_limit() {
     }
 }
 
-/// Cache entries never expire - we explicitly invalidate on mutations.
+/// Cache entries never expire — we use deferred kernel cache invalidation
+/// (via Notifier::inval_entry) after mutations to keep the dcache consistent.
 /// This is safe because we are the only writer to the filesystem.
 const TTL: Duration = Duration::MAX;
 
@@ -602,7 +603,7 @@ impl Filesystem for AgentFSFuse {
     ///
     /// Verifies the target is a directory and is empty before removal.
     /// Returns `ENOTDIR` if not a directory, `ENOTEMPTY` if not empty.
-    fn rmdir(&mut self, _req: &Request, parent: u64, name: &OsStr, reply: ReplyEmpty) {
+    fn rmdir(&mut self, req: &Request, parent: u64, name: &OsStr, reply: ReplyEmpty) {
         tracing::debug!("FUSE::rmdir: parent={}, name={:?}", parent, name);
 
         let Some(name_str) = name.to_str() else {
@@ -619,6 +620,7 @@ impl Filesystem for AgentFSFuse {
         match result {
             Ok(()) => {
                 reply.ok();
+                req.deferred_notifier().inval_entry(parent, name);
             }
             Err(e) => reply.error(error_to_errno(&e)),
         }
@@ -772,7 +774,7 @@ impl Filesystem for AgentFSFuse {
     /// Removes a file (unlinks it from the directory).
     ///
     /// Gets the file's inode before removal to clean up the path cache.
-    fn unlink(&mut self, _req: &Request, parent: u64, name: &OsStr, reply: ReplyEmpty) {
+    fn unlink(&mut self, req: &Request, parent: u64, name: &OsStr, reply: ReplyEmpty) {
         tracing::debug!("FUSE::unlink: parent={}, name={:?}", parent, name);
 
         let Some(name_str) = name.to_str() else {
@@ -789,6 +791,7 @@ impl Filesystem for AgentFSFuse {
         match result {
             Ok(()) => {
                 reply.ok();
+                req.deferred_notifier().inval_entry(parent, name);
             }
             Err(e) => reply.error(error_to_errno(&e)),
         }
@@ -799,7 +802,7 @@ impl Filesystem for AgentFSFuse {
     /// Moves `name` from `parent` to `newname` under `newparent`.
     fn rename(
         &mut self,
-        _req: &Request,
+        req: &Request,
         parent: u64,
         name: &OsStr,
         newparent: u64,
@@ -841,6 +844,9 @@ impl Filesystem for AgentFSFuse {
         match result {
             Ok(()) => {
                 reply.ok();
+                let dn = req.deferred_notifier();
+                dn.inval_entry(parent, name);
+                dn.inval_entry(newparent, newname);
             }
             Err(e) => reply.error(error_to_errno(&e)),
         }
diff --git a/cli/src/fuser/deferred_notify.rs b/cli/src/fuser/deferred_notify.rs
@@ -0,0 +1,43 @@
+use log::debug;
+use std::{
+    ffi::{OsStr, OsString},
+    sync::mpsc,
+};
+
+/// A queued invalidation operation to be flushed by the notify thread.
+#[derive(Debug)]
+pub enum NotifyOp {
+    InvalEntry { parent: u64, name: OsString },
+}
+
+/// Queues kernel cache invalidation requests for deferred execution.
+///
+/// FUSE notification writes to /dev/fuse cannot be issued from the session
+/// loop thread — even outside callbacks — because the kernel processes
+/// FUSE_NOTIFY_INVAL_ENTRY synchronously within the writev() call. That
+/// processing can trigger d_invalidate → iput → FUSE_FORGET, which needs
+/// the daemon to be reading /dev/fuse. Since the session loop thread is
+/// blocked in writev(), it can't read, causing a deadlock.
+///
+/// DeferredNotifier solves this by sending operations over an mpsc channel
+/// to a dedicated background thread that writes to /dev/fuse independently
+/// of the session loop.
+#[derive(Debug, Clone)]
+pub struct DeferredNotifier {
+    tx: mpsc::Sender<NotifyOp>,
+}
+
+impl DeferredNotifier {
+    pub(crate) fn new(tx: mpsc::Sender<NotifyOp>) -> Self {
+        Self { tx }
+    }
+
+    pub fn inval_entry(&self, parent: u64, name: &OsStr) {
+        if let Err(e) = self.tx.send(NotifyOp::InvalEntry {
+            parent,
+            name: name.to_os_string(),
+        }) {
+            debug!("deferred inval_entry send failed (notify thread gone?): {e}");
+        }
+    }
+}
diff --git a/cli/src/fuser/mod.rs b/cli/src/fuser/mod.rs
@@ -51,6 +51,7 @@ mod channel;
     unexpected_cfgs,
     clippy::manual_is_multiple_of
 )]
+pub(crate) mod deferred_notify;
 mod ll;
 #[allow(clippy::io_other_error)]
 mod mnt;
diff --git a/cli/src/fuser/request.rs b/cli/src/fuser/request.rs
@@ -12,6 +12,7 @@ use std::convert::TryInto;
 use std::path::Path;
 
 use super::channel::ChannelSender;
+use super::deferred_notify::DeferredNotifier;
 use super::ll::Request as _;
 use super::reply::ReplyDirectoryPlus;
 use super::reply::{Reply, ReplyDirectory, ReplySender};
@@ -25,6 +26,8 @@ use super::{ll, KernelConfig};
 pub struct Request<'a> {
     /// Channel sender for sending the reply
     ch: ChannelSender,
+    /// Deferred notifier for enqueueing cache invalidations
+    deferred: &'a DeferredNotifier,
     /// Request raw data
     #[allow(unused)]
     data: &'a [u8],
@@ -34,7 +37,11 @@ pub struct Request<'a> {
 
 impl<'a> Request<'a> {
     /// Create a new request from the given data
-    pub(crate) fn new(ch: ChannelSender, data: &'a [u8]) -> Option<Request<'a>> {
+    pub(crate) fn new(
+        ch: ChannelSender,
+        deferred: &'a DeferredNotifier,
+        data: &'a [u8],
+    ) -> Option<Request<'a>> {
         let request = match ll::AnyRequest::try_from(data) {
             Ok(request) => request,
             Err(err) => {
@@ -43,7 +50,16 @@ impl<'a> Request<'a> {
             }
         };
 
-        Some(Self { ch, data, request })
+        Some(Self {
+            ch,
+            deferred,
+            data,
+            request,
+        })
+    }
+
+    pub fn deferred_notifier(&self) -> &DeferredNotifier {
+        self.deferred
     }
 
     /// Dispatch request to the given filesystem.
diff --git a/cli/src/fuser/session.rs b/cli/src/fuser/session.rs
@@ -14,6 +14,9 @@ use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 use std::thread::{self, JoinHandle};
 
+use std::sync::mpsc;
+
+use super::deferred_notify::{DeferredNotifier, NotifyOp};
 use super::ll::fuse_abi as abi;
 use super::request::Request;
 use super::Filesystem;
@@ -64,6 +67,10 @@ pub struct Session<FS: Filesystem> {
     pub(crate) initialized: bool,
     /// True if the filesystem was destroyed (destroy operation done)
     pub(crate) destroyed: bool,
+    /// Sender half of the deferred notification queue
+    notify_tx: Option<mpsc::Sender<NotifyOp>>,
+    /// Receiver half — moved to the notify thread in run()
+    notify_rx: Option<mpsc::Receiver<NotifyOp>>,
 }
 
 impl<FS: Filesystem> AsFd for Session<FS> {
@@ -109,6 +116,8 @@ impl<FS: Filesystem> Session<FS> {
             SessionACL::Owner
         };
 
+        let (notify_tx, notify_rx) = mpsc::channel();
+
         Ok(Session {
             filesystem,
             ch,
@@ -119,13 +128,16 @@ impl<FS: Filesystem> Session<FS> {
             proto_minor: 0,
             initialized: false,
             destroyed: false,
+            notify_tx: Some(notify_tx),
+            notify_rx: Some(notify_rx),
         })
     }
 
     /// Wrap an existing /dev/fuse file descriptor. This doesn't mount the
     /// filesystem anywhere; that must be done separately.
     pub fn from_fd(filesystem: FS, fd: OwnedFd, acl: SessionACL) -> Self {
         let ch = Channel::new(Arc::new(fd.into()));
+        let (notify_tx, notify_rx) = mpsc::channel();
         Session {
             filesystem,
             ch,
@@ -136,6 +148,8 @@ impl<FS: Filesystem> Session<FS> {
             proto_minor: 0,
             initialized: false,
             destroyed: false,
+            notify_tx: Some(notify_tx),
+            notify_rx: Some(notify_rx),
         }
     }
 
@@ -146,20 +160,43 @@ impl<FS: Filesystem> Session<FS> {
     /// # Errors
     /// Returns any final error when the session comes to an end.
     pub fn run(&mut self) -> io::Result<()> {
+        let notify_rx = self.notify_rx.take().expect("run() called more than once");
+        let notifier = self.notifier();
+        let notify_handle = thread::spawn(move || {
+            for op in notify_rx {
+                let res = match op {
+                    NotifyOp::InvalEntry { parent, ref name } => {
+                        notifier.inval_entry(parent, name.as_os_str())
+                    }
+                };
+                if let Err(e) = res {
+                    debug!("FUSE notify failed: {e}");
+                }
+            }
+        });
+
+        // A single DeferredNotifier shared by all requests in this session,
+        // avoiding a Sender clone on every FUSE request dispatch.
+        let deferred =
+            DeferredNotifier::new(self.notify_tx.as_ref().expect("notify_tx missing").clone());
+
         // Buffer for receiving requests from the kernel. Only one is allocated and
         // it is reused immediately after dispatching to conserve memory and allocations.
         let mut buffer = vec![0; BUFFER_SIZE];
         let buf = aligned_sub_buf(&mut buffer, std::mem::align_of::<abi::fuse_in_header>());
+        let mut result = Ok(());
         loop {
             // Read the next request from the given channel to kernel driver
             // The kernel driver makes sure that we get exactly one request per read
             match self.ch.receive(buf) {
-                Ok(size) => match Request::new(self.ch.sender(), &buf[..size]) {
-                    // Dispatch request
-                    Some(req) => req.dispatch(self),
-                    // Quit loop on illegal request
-                    None => break,
-                },
+                Ok(size) => {
+                    match Request::new(self.ch.sender(), &deferred, &buf[..size]) {
+                        // Dispatch request
+                        Some(req) => req.dispatch(self),
+                        // Quit loop on illegal request
+                        None => break,
+                    }
+                }
                 Err(err) => match err.raw_os_error() {
                     Some(
                           ENOENT // Operation interrupted. Accordingly to FUSE, this is safe to retry
@@ -168,11 +205,23 @@ impl<FS: Filesystem> Session<FS> {
                     ) => continue,
                     Some(ENODEV) => break,
                     // Unhandled error
-                    _ => return Err(err),
+                    _ => {
+                        result = Err(err);
+                        break;
+                    }
                 },
             }
         }
-        Ok(())
+
+        // Drop all senders to close the channel, then join the notify thread
+        // to ensure in-flight invalidations are flushed before returning.
+        drop(deferred);
+        self.notify_tx.take();
+        if let Err(e) = notify_handle.join() {
+            warn!("notify thread panicked: {e:?}");
+        }
+
+        result
     }
 
     /// Unmount the filesystem
diff --git a/cli/tests/all.sh b/cli/tests/all.sh
@@ -22,4 +22,5 @@ DIR="$(dirname "$0")"
 "$DIR/test-mount.sh"
 "$DIR/test-overlay-whiteout.sh"
 "$DIR/test-overlay-delta-in-base-dir.sh"
+"$DIR/test-fuse-cache-invalidation.sh"
 "$DIR/test-symlinks.sh" || true  # Requires user namespaces (may fail in CI)
diff --git a/cli/tests/test-fuse-cache-invalidation.sh b/cli/tests/test-fuse-cache-invalidation.sh
