fs: Coalesce queued rescans after watcher overflow (#60098)

After a filesystem watcher loses sync (e.g. a `git pull` that changes
many files overflows the backend's event queue), the backend can enqueue
many `Rescan` markers in quick succession. The dispatch thread processed
each one separately, and each rescan invokes every registration for that
watcher mode, so a single burst could kick off several full worktree
scans at once — leading to sustained CPU and an unresponsive project
panel.

This adds `dispatch_batch`, which handles the first event and drains the
events already waiting in the channel, forwarding at most one `Rescan`
per `WatcherMode` per drained batch while letting ordinary filesystem
events and errors pass through untouched. A later watcher overflow can
still trigger another recovery scan.

Reported with a reproduction and patch in #59610.

Release Notes:

- Batch file watcher rescan events to improve Zed's responsiveness under
heavy FS usage

crates/fs/src/fs_watcher.rs

@@ -931,6 +931,34 @@ impl GlobalWatcher {
         }
     }
 
+    fn dispatch_batch(
+        &self,
+        first: DispatchEvent,
+        event_rx: &async_channel::Receiver<DispatchEvent>,
+    ) {
+        // A single backend overflow can enqueue many rescan markers. One rescan
+        // per mode covers the entire drained batch; ordinary events still run.
+        let mut native_rescan_dispatched = false;
+        let mut poll_rescan_dispatched = false;
+
+        for (mode, event) in
+            std::iter::once(first).chain(std::iter::from_fn(|| event_rx.try_recv().ok()))
+        {
+            let rescan_dispatched = match mode {
+                WatcherMode::Native => &mut native_rescan_dispatched,
+                WatcherMode::Poll => &mut poll_rescan_dispatched,
+            };
+            if event.as_ref().is_ok_and(notify::Event::need_rescan) {
+                if *rescan_dispatched {
+                    continue;
+                }
+                *rescan_dispatched = true;
+            }
+
+            self.dispatch(mode, event);
+        }
+    }
+
     fn start_native_watch_limit_cooldown(&self, path: &Path) {
         let mut state = self.state.lock();
         let now = Instant::now();
@@ -1078,8 +1106,8 @@ fn global_watcher() -> &'static GlobalWatcher {
         std::thread::Builder::new()
             .name("fs-watcher-dispatch".to_owned())
             .spawn(move || {
-                while let Ok((mode, event)) = event_rx.recv_blocking() {
-                    global_watcher().dispatch(mode, event);
+                while let Ok(first) = event_rx.recv_blocking() {
+                    global_watcher().dispatch_batch(first, &event_rx);
                 }
             })
             .expect("failed to spawn fs watcher dispatch thread");
@@ -1466,6 +1494,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn queued_rescans_are_coalesced_without_dropping_normal_events() {
+        let (watcher, fired) = recording_watcher();
+        let (event_tx, event_rx) = async_channel::unbounded();
+        let rescan = || notify::Event::new(EventKind::Other).set_flag(notify::event::Flag::Rescan);
+
+        event_tx
+            .try_send((WatcherMode::Native, Ok(rescan())))
+            .unwrap();
+        event_tx
+            .try_send((WatcherMode::Native, Ok(modify_event("/repo/a/file.txt"))))
+            .unwrap();
+        watcher.dispatch_batch((WatcherMode::Native, Ok(rescan())), &event_rx);
+
+        let mut got = fired.lock().clone();
+        got.sort();
+        assert_eq!(
+            got,
+            vec![
+                "/repo/a".to_owned(),
+                "/repo/a".to_owned(),
+                "/repo/a/nested".to_owned(),
+                "/repo/b".to_owned(),
+            ]
+        );
+    }
+
     #[test]
     #[cfg(target_os = "windows")]
     fn event_dispatches_when_reported_path_has_verbatim_prefix() {