fixed concurrency bugs + performance

crates/pdfs-fuse/src/background.rs

@@ -43,7 +43,7 @@ fn forget_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid)
 /// read-side twin of `docs/BUGS.md` B74. Inode numbers are per-mount, so each
 /// mount must be notified through its **own** channel; that pairing is what
 /// [`Core::for_each_mount`] exists to preserve.
-fn apply_event(core: &Core, event: &DriveEvent) {
+fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) {
     match event {
         DriveEvent::NodeUpdated {
             node_uid,
@@ -56,9 +56,16 @@ fn apply_event(core: &Core, event: &DriveEvent) {
             // create, and re-fetching would replace the size and mtime of the
             // write we just accepted with the stale revision's — making a file
             // that was copied in seconds ago read as empty until its upload
-            // lands (offline.md Phase 3). The parent listing is still refreshed
-            // below; it is only the node itself that is ours to define.
-            let ours = !*is_trashed && core.pending.lock().contains_key(node_uid);
+            // lands (offline.md Phase 3).
+            //
+            // A write that has already drained is the same story one step later:
+            // the drain brought the tree level with the revision it sealed, so
+            // the feed's report of that revision has nothing to add, and acting
+            // on it evicts the very bytes we uploaded. Claimed unconditionally,
+            // even for a trash, so an echo we no longer care about does not
+            // linger to suppress a later foreign change.
+            let echo = core.take_self_change(node_uid);
+            let ours = !*is_trashed && (echo || core.pending.lock().contains_key(node_uid));
             if ours {
                 debug!(uid = %node_uid, "ignoring remote event for a node with a queued write");
             }
@@ -77,22 +84,24 @@ fn apply_event(core: &Core, event: &DriveEvent) {
                         let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
                     }
                 }
-                // A create (or move-in) shows up as a change to the parent
-                // listing; drop it so the new child is picked up on the next
-                // readdir.
-                if let Some(parent_uid) = parent_node_uid
-                    && let Some(&parent) = st.by_uid.get(parent_uid)
-                {
-                    st.invalidate_listing(parent);
-                    if let Some(notifier) = notifier {
-                        let _ = notifier.inval_inode(INodeNo(parent), 0, 0);
-                    }
-                }
             });
             // Once, not once per mount: the content cache is per-daemon.
             if had_node {
                 core.cache.evict(node_uid);
             }
+            // A create, rename or move-in shows up as a change to the parent
+            // listing too, which is recorded for the end of the batch rather
+            // than acted on here — see [`DirtyParents`].
+            if let Some(parent_uid) = parent_node_uid {
+                match ours {
+                    true => dirty
+                        .ours
+                        .entry(parent_uid.clone())
+                        .or_default()
+                        .insert(node_uid.clone()),
+                    false => dirty.foreign.insert(parent_uid.clone()),
+                };
+            }
         }
         DriveEvent::NodeDeleted { node_uid, .. } => {
             core.cache.evict(node_uid);
@@ -121,6 +130,88 @@ fn apply_event(core: &Core, event: &DriveEvent) {
     }
 }
 
+/// The folders one batch of events changed, collected so the listings are
+/// dropped once at the end of the batch rather than once per event.
+///
+/// Both halves of that matter. A burst of uploads produces one event per file
+/// all naming the *same* folder, and dropping the listing per event made every
+/// interleaved lookup re-enumerate that folder from the API — quadratic in its
+/// size, and slow enough to stall a busy mount. Splitting foreign changes from
+/// our own then removes the re-enumeration entirely for the common case: a file
+/// this daemon created is already in the tree under the right parent, so there
+/// is nothing for a re-enumeration to discover.
+///
+/// Neither is a weakening. The state after the batch is what it always was.
+#[derive(Default)]
+struct DirtyParents {
+    /// Folders changed by someone else. Their listings must go: an event names
+    /// no name, so a rename inside a folder is indistinguishable from a write to
+    /// a file in it, and only re-enumerating can tell us which happened.
+    foreign: HashSet<NodeUid>,
+    /// Folders changed by this daemon, and the children the changes were about.
+    /// Checked per mount — a mount that already lists every one of those
+    /// children under that parent keeps its listing.
+    ours: HashMap<NodeUid, HashSet<NodeUid>>,
+}
+
+/// Whether this mount's cached listing of `parent_uid` already accounts for
+/// `child_uid`, so a change we made to that child leaves nothing to re-read.
+///
+/// "Nothing cached" counts as current: an absent listing is re-enumerated on the
+/// next `readdir` regardless, so there is nothing to drop.
+fn listing_accounts_for(st: &State, parent_uid: &NodeUid, child_uid: &NodeUid) -> bool {
+    let Some(&parent) = st.by_uid.get(parent_uid) else {
+        return true;
+    };
+    let Some(children) = st.children.get(&parent) else {
+        return true;
+    };
+    let Some(&child) = st.by_uid.get(child_uid) else {
+        return false;
+    };
+    children.contains(&child) && st.entries.get(&child).is_some_and(|e| e.parent == parent)
+}
+
+/// Drop the cached listing of every folder a batch of events changed, in every
+/// mount that holds one, so the next `readdir`/`lookup` re-enumerates it.
+///
+/// Runs strictly after the whole batch is applied: doing it per event would
+/// leave a window where a concurrent `readdir` re-populates a listing that a
+/// later event in the same batch then has to drop again.
+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);
+            }
+        };
+        for parent_uid in &dirty.foreign {
+            drop_listing(st, parent_uid);
+        }
+        for (parent_uid, children) in &dirty.ours {
+            if dirty.foreign.contains(parent_uid) {
+                continue;
+            }
+            // One child this mount has not placed is enough: the listing it is
+            // serving is missing something the user just made.
+            if children
+                .iter()
+                .all(|child| listing_accounts_for(st, parent_uid, child))
+            {
+                continue;
+            }
+            drop_listing(st, parent_uid);
+        }
+    });
+}
+
 /// Poll the remote event cursor forever, applying each batch to the shared
 /// state. Resumes from the cursor persisted in the DB so changes made while
 /// unmounted are applied; only a first-ever mount seeds from the server head.
@@ -190,6 +281,11 @@ pub(super) async fn run_event_sync(
             continue;
         }
         debug!(count = events.len(), "applying remote events");
+        // Folders whose listing this batch invalidates, applied once at the end
+        // rather than per event — see [`DirtyParents`]. Every `break` below falls
+        // through to that flush, so an interrupted batch still publishes what it
+        // did apply.
+        let mut dirty = DirtyParents::default();
         for event in &events {
             // Converge the SDK's own caches (folder keys, entity cache) on the
             // server before applying the event to our tree. Without this, a node
@@ -200,7 +296,7 @@ pub(super) async fn run_event_sync(
                 warn!(error = %e, "sdk cache invalidation for event failed; retaining cursor for retry");
                 break;
             }
-            apply_event(&core, event);
+            apply_event(&core, event, &mut dirty);
             let applied = event.id().clone();
             if let Err(e) = db.set_event_cursor(applied.as_str()) {
                 warn!(error = %e, "persist event cursor failed; retaining prior cursor for retry");
@@ -211,6 +307,7 @@ pub(super) async fn run_event_sync(
             // that was not already made visible locally.
             cursor = Some(applied);
         }
+        flush_dirty_parents(&core, &dirty);
     }
 }
 

crates/pdfs-fuse/src/drain.rs

@@ -229,6 +229,7 @@ impl Core {
             }
         }
         self.db.delete_op(op.id)?;
+        self.note_self_change(&uid);
         info!(%uid, name = %landed, "pending rename landed");
         Ok(())
     }
@@ -344,6 +345,9 @@ impl Core {
         // sync, whereas a surviving op would create the file a second time.
         self.db.delete_op(op.id)?;
         self.adopt_real_uid(&local, &real)?;
+        // The feed will report this create back to us; the tree already has it
+        // under its real uid, so that event is ours to ignore (`Core::self_changes`).
+        self.note_self_change(&real);
         if let Some(blob) = op.blob_path.as_deref() {
             self.cache.discard_staged(Path::new(blob));
         }
@@ -845,6 +849,11 @@ impl Core {
                 return;
             }
         };
+        // This function's whole job is to bring the tree level with the revision
+        // we just sealed, so the feed's report of that revision has nothing left
+        // to tell us. Claimed here rather than at the upload call so it is only
+        // recorded when the node was actually re-read (`Core::self_changes`).
+        self.note_self_change(uid);
         // Rebase any *open* write handles targeting this node. A handle opened
         // before this upload carries a stale base — its `base_mtime`/`base_size`
         // name the revision we just replaced. Without this update,

crates/pdfs-fuse/src/lib.rs

@@ -320,6 +320,26 @@ struct Core {
     /// uploading that same tree — the engine would upload files as they vanish
     /// and then walk the FUSE mount as if it were local.
     sync_locks: Arc<Mutex<HashMap<i64, Arc<Mutex<()>>>>>,
+    /// Nodes this daemon changed on the remote itself, and how many echoes of
+    /// those changes the event feed still owes us.
+    ///
+    /// The feed replays our own writes back at us seconds later, and a
+    /// `NodeUpdated` carries no revision id — nothing in it distinguishes "another
+    /// device changed this file" from "you changed this file". Treating the echo
+    /// as foreign is not free: it evicts the content blob we just uploaded, so the
+    /// next read re-downloads a revision the API may not be serving yet. That is
+    /// how a SQLite database written on the mount came back `malformed`, and how a
+    /// mount doing bulk uploads spent its time re-fetching its own bytes.
+    ///
+    /// So the first event for a node we changed ourselves, within
+    /// [`SELF_CHANGE_TTL_MS`], is consumed rather than applied: the tree already
+    /// holds what that event describes, because [`Core::refresh_after_upload`] put
+    /// it there. Anything beyond that applies normally, which bounds the staleness
+    /// a genuine remote change can suffer to one event.
+    ///
+    /// Shared across forks like `pending`: the drain that records the change and
+    /// the event task that consumes the echo are both per-daemon.
+    self_changes: Arc<Mutex<HashMap<NodeUid, SelfChange>>>,
     /// Every live inode space: this mount's own `state` plus one per on-demand
     /// fork ([`Core::fork_state`]). Shared by every fork, unlike `state` itself.
     ///
@@ -337,6 +357,58 @@ struct Core {
     states: Arc<StateRegistry>,
 }
 
+/// How long a change this daemon made stays recognisable as its own. Generous
+/// against a feed that can lag, but finite: past it, an event for the node is
+/// applied like any other. See [`Core::self_changes`].
+const SELF_CHANGE_TTL_MS: i64 = 120_000;
+
+/// A remote change this daemon made, awaiting its echo from the event feed.
+struct SelfChange {
+    /// When we made it, in the same epoch millis as [`now_millis`].
+    at_ms: i64,
+    /// Echoes not yet seen. More than one because a single file can be created
+    /// and then have a revision uploaded, and the feed reports both.
+    echoes: u32,
+}
+
+/// Record one remote change this daemon made. Split from [`Core`] so the
+/// expiry and echo-counting rules can be tested against a clock the test owns.
+fn note_self_change(changes: &mut HashMap<NodeUid, SelfChange>, uid: &NodeUid, now_ms: i64) {
+    // Echoes that never arrived — a node trashed before its event came round, a
+    // feed that skipped it — would otherwise accumulate for the life of the
+    // daemon. Pruning on write keeps the map the size of the recent queue.
+    changes.retain(|_, c| now_ms - c.at_ms < SELF_CHANGE_TTL_MS);
+    let change = changes.entry(uid.clone()).or_insert(SelfChange {
+        at_ms: now_ms,
+        echoes: 0,
+    });
+    change.at_ms = now_ms;
+    change.echoes += 1;
+}
+
+/// Claim one echo, per [`Core::take_self_change`].
+fn take_self_change(
+    changes: &mut HashMap<NodeUid, SelfChange>,
+    uid: &NodeUid,
+    now_ms: i64,
+) -> bool {
+    let Some(change) = changes.get_mut(uid) else {
+        return false;
+    };
+    // Too old to attribute: the feed is far enough behind that this is as likely
+    // to be someone else's change as ours, and guessing wrong here costs the user
+    // a stale file.
+    if now_ms - change.at_ms >= SELF_CHANGE_TTL_MS {
+        changes.remove(uid);
+        return false;
+    }
+    change.echoes -= 1;
+    if change.echoes == 0 {
+        changes.remove(uid);
+    }
+    true
+}
+
 /// One live inode space, as published to the shared registry.
 ///
 /// The notifier travels with the state because the two are only meaningful
@@ -436,6 +508,21 @@ impl Core {
         }
     }
 
+    /// Record that this daemon just changed `uid` on the remote, so the event
+    /// feed's echo of that change is recognised instead of re-applied. Called
+    /// from the drain once the remote has actually accepted the change and the
+    /// tree has been brought level with it. See [`Core::self_changes`].
+    pub(crate) fn note_self_change(&self, uid: &NodeUid) {
+        note_self_change(&mut self.self_changes.lock(), uid, now_millis());
+    }
+
+    /// Claim one outstanding echo for `uid`, reporting whether this event is one
+    /// this daemon caused. Consuming rather than merely testing is what bounds
+    /// the suppression: the next event for the node applies normally.
+    fn take_self_change(&self, uid: &NodeUid) -> bool {
+        take_self_change(&mut self.self_changes.lock(), uid, now_millis())
+    }
+
     /// Rehydrate the in-memory `State` maps from the DB on mount, so a cold
     /// start serves previously-seen metadata (stable inodes, instant listings)
     /// without re-hitting the API. The root inode is already installed by
@@ -3548,8 +3635,9 @@ mod pending_size_tests {
 #[cfg(test)]
 mod tests {
     use super::{
-        Intervals, PendingRevision, StateRegistry, conflict_name, copy_pending_for_truncate,
-        fuse_name, is_stale_mount, node_visible, rename_needs_queue,
+        HashMap, Intervals, PendingRevision, SELF_CHANGE_TTL_MS, StateRegistry, conflict_name,
+        copy_pending_for_truncate, fuse_name, is_stale_mount, node_visible, note_self_change,
+        parse_node_uid, rename_needs_queue, take_self_change,
     };
     use pdfs_core::cache::{Baseline, StagedWrite};
 
@@ -3600,6 +3688,67 @@ mod tests {
         );
     }
 
+    /// An event feed carries no revision id, so the only thing that tells a
+    /// remote change apart from the echo of one this daemon just made is this
+    /// bookkeeping. Getting it wrong is expensive in both directions: too eager
+    /// and we evict the bytes we uploaded and re-download them from an API that
+    /// may still be serving the previous revision (a SQLite file on the mount
+    /// read back `malformed`); too generous and a real change from another device
+    /// is ignored.
+    #[test]
+    fn a_self_change_is_claimed_once_and_expires() {
+        let mut changes = HashMap::new();
+        let uid = parse_node_uid("vol~link").unwrap();
+        let other = parse_node_uid("vol~other").unwrap();
+        let t0 = 1_000_000;
+
+        assert!(
+            !take_self_change(&mut changes, &uid, t0),
+            "a node we never touched is never ours"
+        );
+
+        note_self_change(&mut changes, &uid, t0);
+        assert!(
+            take_self_change(&mut changes, &uid, t0 + 5_000),
+            "the echo of our own change is claimed"
+        );
+        assert!(
+            !take_self_change(&mut changes, &uid, t0 + 6_000),
+            "and only once, so a later foreign change still applies"
+        );
+
+        // A create followed by a revision upload is two changes to one node, and
+        // the feed reports both.
+        note_self_change(&mut changes, &uid, t0);
+        note_self_change(&mut changes, &uid, t0);
+        assert!(take_self_change(&mut changes, &uid, t0 + 1));
+        assert!(take_self_change(&mut changes, &uid, t0 + 2));
+        assert!(!take_self_change(&mut changes, &uid, t0 + 3));
+
+        // Past the window the attribution is a guess, and a wrong guess leaves
+        // the user looking at a stale file.
+        note_self_change(&mut changes, &uid, t0);
+        assert!(!take_self_change(
+            &mut changes,
+            &uid,
+            t0 + SELF_CHANGE_TTL_MS
+        ));
+        assert!(
+            !changes.contains_key(&uid),
+            "an expired change is dropped, not left to be re-tested"
+        );
+
+        // An echo that never arrives must not accumulate for the life of the
+        // daemon; the next recorded change prunes it.
+        note_self_change(&mut changes, &uid, t0);
+        note_self_change(&mut changes, &other, t0 + SELF_CHANGE_TTL_MS);
+        assert_eq!(
+            changes.len(),
+            1,
+            "recording a change prunes the ones whose echo never came"
+        );
+    }
+
     /// The predicate must answer *only* for a dead FUSE connection. A healthy
     /// directory and an absent path are both "not stale" — widening it to any
     /// `metadata` error would make the on-demand restore lazily unmount paths

crates/pdfs-fuse/src/mount.rs

@@ -212,6 +212,7 @@ pub fn mount(
         conflict_notified: Arc::new(Mutex::new(HashSet::new())),
         sweep_mode,
         own_sealed_revs: Arc::new(Mutex::new(HashMap::new())),
+        self_changes: Arc::new(Mutex::new(HashMap::new())),
         thumb_gen: Arc::new(Mutex::new(HashSet::new())),
         no_thumbnail: Arc::new(Mutex::new(HashMap::new())),
         size_upgrades: Arc::new(Mutex::new(HashMap::new())),