cli: Fix FUSE kernel cache serving stale directory listings

broady · broady · commit 26e89c20e078 · 2026-02-09T01:28:02.000-05:00
When readdirplus populates the kernel dcache with TTL=MAX, the
kernel caches those entries indefinitely. After unlink/rmdir/
rename the kernel drops the specific entry it operated on, but
the parent directory's cached readdirplus result can still contain
stale entries on subsequent ls calls.

This was most visible in overlay mode: after whiteout-deleting a
base file, ls still showed the file because the kernel served
the cached readdirplus result without re-querying userspace.

Fix by sending Notifier::inval_entry to the kernel after unlink,
rmdir, and rename so the dcache drops the affected entry. These
notifications cannot be issued from inside FUSE callbacks because
the kernel may synchronously respond with FUSE_FORGET, which the
single-threaded session loop cannot process until the current
callback returns -- deadlocking the mount.

Instead, callbacks enqueue invalidation requests into a
DeferredNotifier (mpsc channel) that a background thread drains
and writes to /dev/fuse. The session loop remains free to read
FUSE_FORGET while the notification thread writes.

Add test-fuse-cache-invalidation.sh 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,39 @@
+use log::debug;
+use std::{
+    ffi::{OsStr, OsString},
+    sync::mpsc,
+};
+
+/// A queued invalidation operation to be flushed between FUSE callbacks.
+#[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 within
+/// filesystem callbacks because the kernel may respond with FUSE_FORGET,
+/// which the single-threaded session loop cannot process until the current
+/// callback returns — causing a deadlock. DeferredNotifier solves this by
+/// enqueueing operations that a dedicated background thread flushes
+/// asynchronously.
+#[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
