worktree: Don't eagerly remove watchers (#58692)

Self-Review Checklist:

- [x] I've reviewed my own diff for quality, security, and reliability
- [x] Unsafe blocks (if any) have justifying comments
- [x] The content is consistent with the [UI/UX
checklist](https://github.com/zed-industries/zed/blob/main/CONTRIBUTING.md#uiux-checklist)
- [x] Tests cover the new/changed behavior
- [x] Performance impact has been considered and is acceptable

Closes #ISSUE

Release Notes:

- N/A or Added/Fixed/Improved ...

crates/fs/src/fs.rs

@@ -1335,6 +1335,39 @@ struct FakeFsState {
     moves: std::collections::HashMap<u64, PathBuf>,
     job_event_subscribers: Arc<Mutex<Vec<JobEventSender>>>,
     trash: Vec<(TrashedEntry, FakeFsEntry)>,
+    file_to_create_before_watch_add: Option<(PathBuf, PathBuf)>,
+}
+
+#[cfg(feature = "test-support")]
+impl FakeFsState {
+    fn create_file_before_watch_add(&mut self, watch_path: &Path) -> Result<()> {
+        let Some((pending_watch_path, file_path)) = self.file_to_create_before_watch_add.take()
+        else {
+            return Ok(());
+        };
+        if pending_watch_path != watch_path {
+            self.file_to_create_before_watch_add = Some((pending_watch_path, file_path));
+            return Ok(());
+        }
+
+        let inode = self.get_and_increment_inode();
+        let mtime = self.get_and_increment_mtime();
+        self.write_path(&file_path, |entry| {
+            let btree_map::Entry::Vacant(entry) = entry else {
+                anyhow::bail!("file already exists: {}", file_path.display());
+            };
+            entry.insert(FakeFsEntry::File {
+                inode,
+                mtime,
+                len: 0,
+                content: Vec::new(),
+                git_dir_path: None,
+            });
+            Ok(())
+        })?;
+        self.emit_event([(file_path, Some(PathEventKind::Created))]);
+        Ok(())
+    }
 }
 
 #[cfg(feature = "test-support")]
@@ -1621,6 +1654,7 @@ impl FakeFs {
                 moves: Default::default(),
                 job_event_subscribers: Arc::new(Mutex::new(Vec::new())),
                 trash: Vec::new(),
+                file_to_create_before_watch_add: None,
             })),
         });
 
@@ -1796,6 +1830,17 @@ impl FakeFs {
         self.state.lock().buffered_events.clear();
     }
 
+    pub fn create_file_before_next_watch_add(
+        &self,
+        watch_path: impl AsRef<Path>,
+        path: impl AsRef<Path>,
+    ) {
+        self.state.lock().file_to_create_before_watch_add = Some((
+            normalize_path(watch_path.as_ref()),
+            normalize_path(path.as_ref()),
+        ));
+    }
+
     pub fn flush_events(&self, count: usize) {
         self.state.lock().flush_events(count);
     }
@@ -2591,7 +2636,6 @@ impl FakeFsEntry {
 #[cfg(feature = "test-support")]
 struct FakeWatcher {
     tx: async_channel::Sender<Vec<PathEvent>>,
-    original_path: PathBuf,
     fs_state: Arc<Mutex<FakeFsState>>,
     prefixes: Mutex<Vec<PathBuf>>,
 }
@@ -2599,19 +2643,34 @@ struct FakeWatcher {
 #[cfg(feature = "test-support")]
 impl Watcher for FakeWatcher {
     fn add(&self, path: &Path) -> Result<()> {
-        if path.starts_with(&self.original_path) {
+        let path = normalize_path(path);
+        self.fs_state
+            .try_lock()
+            .unwrap()
+            .create_file_before_watch_add(&path)?;
+
+        let mut prefixes = self.prefixes.lock();
+        if prefixes.iter().any(|prefix| path.starts_with(prefix)) {
             return Ok(());
         }
+
         self.fs_state
             .try_lock()
             .unwrap()
             .event_txs
-            .push((path.to_owned(), self.tx.clone()));
-        self.prefixes.lock().push(path.to_owned());
+            .push((path.clone(), self.tx.clone()));
+        prefixes.push(path);
         Ok(())
     }
 
-    fn remove(&self, _: &Path) -> Result<()> {
+    fn remove(&self, path: &Path) -> Result<()> {
+        let path = normalize_path(path);
+        self.prefixes.lock().retain(|prefix| prefix != &path);
+        self.fs_state
+            .try_lock()
+            .unwrap()
+            .event_txs
+            .retain(|(watched_path, _)| watched_path != &path);
         Ok(())
     }
 }
@@ -3065,7 +3124,6 @@ impl Fs for FakeFs {
         let executor = self.executor.clone();
         let watcher = Arc::new(FakeWatcher {
             tx,
-            original_path: path.to_owned(),
             fs_state: self.state.clone(),
             prefixes: Mutex::new(vec![path]),
         });

crates/worktree/src/worktree.rs

@@ -3142,7 +3142,34 @@ impl BackgroundScannerState {
         self.snapshot.check_invariants(false);
     }
 
-    fn remove_path(&mut self, path: &RelPath, watcher: &dyn Watcher) {
+    fn remove_path_from_snapshot_and_unwatch(&mut self, path: &RelPath, watcher: &dyn Watcher) {
+        let removed_descendant_abs_paths = self.remove_path_from_snapshot(path);
+        self.unwatch_path(watcher, path, removed_descendant_abs_paths);
+    }
+
+    fn unwatch_path(
+        &mut self,
+        watcher: &dyn Watcher,
+        path: &RelPath,
+        removed_descendant_abs_paths: Vec<PathBuf>,
+    ) {
+        for removed_dir_abs_path in removed_descendant_abs_paths {
+            watcher.remove(&removed_dir_abs_path).log_err();
+        }
+
+        self.snapshot
+            .external_canonical_to_relative
+            .retain(|canonical, relative| {
+                if relative.starts_with(path) {
+                    watcher.remove(canonical.as_ref()).log_err();
+                    false
+                } else {
+                    true
+                }
+            });
+    }
+
+    fn remove_path_from_snapshot(&mut self, path: &RelPath) -> Vec<PathBuf> {
         log::trace!("background scanner removing path {path:?}");
         let mut new_entries;
         let removed_entries;
@@ -3204,23 +3231,10 @@ impl BackgroundScannerState {
             .git_repositories
             .retain(|id, _| removed_ids.binary_search(id).is_err());
 
-        for removed_dir_abs_path in removed_dir_abs_paths {
-            watcher.remove(&removed_dir_abs_path).log_err();
-        }
-
-        self.snapshot
-            .external_canonical_to_relative
-            .retain(|canonical, relative| {
-                if relative.starts_with(path) {
-                    watcher.remove(canonical.as_ref()).log_err();
-                    false
-                } else {
-                    true
-                }
-            });
-
         #[cfg(feature = "test-support")]
         self.snapshot.check_invariants(false);
+
+        return removed_dir_abs_paths;
     }
 
     async fn insert_git_repository(
@@ -4909,10 +4923,11 @@ impl BackgroundScanner {
 
             if self.settings.is_path_excluded(&child_path) {
                 log::debug!("skipping excluded child entry {child_path:?}");
+
                 self.state
                     .lock()
                     .await
-                    .remove_path(&child_path, self.watcher.as_ref());
+                    .remove_path_from_snapshot_and_unwatch(&child_path, self.watcher.as_ref());
                 continue;
             }
 
@@ -5154,13 +5169,18 @@ impl BackgroundScanner {
         // Remove any entries for paths that no longer exist or are being recursively
         // refreshed. Do this before adding any new entries, so that renames can be
         // detected regardless of the order of the paths.
+        let mut paths_to_process = Vec::with_capacity(relative_paths.len());
         for (path, metadata) in relative_paths.iter().zip(metadata.iter()) {
-            if matches!(metadata, Ok(None)) || doing_recursive_update {
-                state.remove_path(path, self.watcher.as_ref());
-            }
+            let removed_descendant_paths = if matches!(metadata, Ok(None)) || doing_recursive_update
+            {
+                state.remove_path_from_snapshot(path)
+            } else {
+                Vec::new()
+            };
+            paths_to_process.push((path, metadata, removed_descendant_paths));
         }
 
-        for (path, metadata) in relative_paths.iter().zip(metadata) {
+        for (path, metadata, removed_descendant_abs_paths) in paths_to_process {
             let abs_path: Arc<Path> = root_abs_path.join(path.as_std_path()).into();
             match metadata {
                 Ok(Some((metadata, canonical_path))) => {
@@ -5242,9 +5262,11 @@ impl BackgroundScanner {
                 }
                 Ok(None) => {
                     self.remove_repo_path(path.clone(), &mut state.snapshot);
+                    state.unwatch_path(self.watcher.as_ref(), path, removed_descendant_abs_paths);
                 }
                 Err(err) => {
                     log::error!("error reading file {abs_path:?} on event: {err:#}");
+                    state.unwatch_path(self.watcher.as_ref(), path, removed_descendant_abs_paths);
                 }
             }
         }

crates/worktree/tests/integration/worktree_tests.rs

@@ -1324,6 +1324,41 @@ async fn test_root_rescan_reconciles_stale_state(cx: &mut TestAppContext) {
     });
 }
 
+#[gpui::test]
+async fn test_root_rescan_does_not_miss_event_before_readding_root_watcher(
+    cx: &mut TestAppContext,
+) {
+    init_test(cx);
+    let fs = FakeFs::new(cx.background_executor.clone());
+    fs.insert_tree("/root", json!({})).await;
+
+    let tree = Worktree::local(
+        Path::new("/root"),
+        true,
+        fs.clone(),
+        Default::default(),
+        true,
+        WorktreeId::from_proto(0),
+        &mut cx.to_async(),
+    )
+    .await
+    .unwrap();
+
+    cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
+        .await;
+
+    fs.create_file_before_next_watch_add("/root", "/root/created-before-root-readd.txt");
+    fs.emit_fs_event("/root", Some(PathEventKind::Rescan));
+
+    wait_for_condition(cx, |cx| {
+        tree.read_with(cx, |tree, _| {
+            tree.entry_for_path(rel_path("created-before-root-readd.txt"))
+                .is_some()
+        })
+    })
+    .await;
+}
+
 #[gpui::test]
 async fn test_subtree_rescan_reports_unchanged_descendants_as_updated(cx: &mut TestAppContext) {
     init_test(cx);