diff --git a/crates/fs/src/fs_watcher.rs b/crates/fs/src/fs_watcher.rs index 82aeaf2a73..98184da38e 100644 --- a/crates/fs/src/fs_watcher.rs +++ b/crates/fs/src/fs_watcher.rs @@ -521,19 +521,19 @@ pub struct WatcherRegistrationId(u32); struct WatcherRegistrationState { callback: Arc, - path: Arc, + path: Arc, mode: WatcherMode, } struct PathRegistrationState { - count: u32, + watcher_ids: Vec, has_os_watcher: bool, } struct WatcherState { watchers: HashMap, - native_path_registrations: HashMap, PathRegistrationState>, - poll_path_registrations: HashMap, PathRegistrationState>, + native_path_registrations: HashMap, PathRegistrationState>, + poll_path_registrations: HashMap, PathRegistrationState>, cooldown_until: Option, last_registration: WatcherRegistrationId, } @@ -547,7 +547,7 @@ impl WatcherState { fn path_registrations( &mut self, mode: WatcherMode, - ) -> &mut HashMap, PathRegistrationState> { + ) -> &mut HashMap, PathRegistrationState> { match mode { WatcherMode::Native => &mut self.native_path_registrations, WatcherMode::Poll => &mut self.poll_path_registrations, @@ -557,16 +557,16 @@ impl WatcherState { fn remove_registration( &mut self, id: WatcherRegistrationId, - ) -> Option<(Arc, WatcherMode)> { + ) -> Option<(Arc, WatcherMode)> { let registration_state = self.watchers.remove(&id)?; let path_registrations = self.path_registrations(registration_state.mode); - let count = path_registrations.get_mut(®istration_state.path)?; - count.count -= 1; - if count.count != 0 { + let path_state = path_registrations.get_mut(®istration_state.path)?; + path_state.watcher_ids.retain(|&existing| existing != id); + if !path_state.watcher_ids.is_empty() { return None; } - let was_actually_watched = count.has_os_watcher; + let was_actually_watched = path_state.has_os_watcher; path_registrations.remove(®istration_state.path); was_actually_watched.then_some((registration_state.path, registration_state.mode)) @@ -588,6 +588,8 @@ impl WatchBackend for T { } } +type DispatchEvent = (WatcherMode, Result); + pub struct GlobalWatcher { state: Mutex, @@ -595,6 +597,7 @@ pub struct GlobalWatcher { // two mutexes because calling watcher.add triggers watcher.event, which needs watchers. native_watcher: Mutex>>, poll_watcher: Mutex>>, + event_tx: async_channel::Sender, } impl GlobalWatcher { @@ -605,6 +608,7 @@ impl GlobalWatcher { mode: WatcherMode, cb: impl Fn(¬ify::Event) + Send + Sync + 'static, ) -> anyhow::Result> { + let path = SanitizedPath::from_arc(path); let mut state = self.state.lock(); let (path_already_covered, path_already_registered) = { let registrations_for_mode = state.path_registrations(mode); @@ -620,10 +624,10 @@ impl GlobalWatcher { } drop(state); - match self.watch(&path, mode) { + match self.watch(path.as_path(), mode) { Ok(()) => {} Err(error) if mode == WatcherMode::Native && is_max_files_watch_error(&error) => { - self.start_native_watch_limit_cooldown(&path); + self.start_native_watch_limit_cooldown(path.as_path()); return Ok(None); } Err(error) => return Err(error), @@ -643,15 +647,85 @@ impl GlobalWatcher { state .path_registrations(mode) .entry(path) - .and_modify(|registration| registration.count += 1) - .or_insert(PathRegistrationState { - count: 1, + .and_modify(|registration| registration.watcher_ids.push(id)) + .or_insert_with(|| PathRegistrationState { + watcher_ids: vec![id], has_os_watcher: !path_already_covered, }); Ok(Some(id)) } + fn enqueue(&self, mode: WatcherMode, event: Result) { + if matches!( + event, + Ok(Event { + kind: EventKind::Access(_), + .. + }) + ) { + return; + } + + // A failed send only happens once the dispatch thread has shut down, at + // which point there's nothing left to dispatch to. + self.event_tx.try_send((mode, event)).ok(); + } + + fn dispatch(&self, mode: WatcherMode, event: Result) { + let event = match event { + Ok(event) => event, + Err(error) => { + log::warn!("watcher error for {mode:?}: {error}"); + return; + } + }; + + log::trace!("global handle event for {mode:?}: {event:?}"); + + let callbacks = { + let state = self.state.lock(); + if event.need_rescan() { + let callbacks = state + .watchers + .values() + .filter(|registration| registration.mode == mode) + .map(|registration| registration.callback.clone()) + .collect::>(); + log::warn!( + "filesystem watcher lost sync for {mode:?}; scheduling rescans for {} registrations", + callbacks.len() + ); + callbacks + } else { + let path_registrations = match mode { + WatcherMode::Native => &state.native_path_registrations, + WatcherMode::Poll => &state.poll_path_registrations, + }; + let mut ids = Vec::new(); + for path in &event.paths { + let sanitized = SanitizedPath::new(path); + for ancestor in sanitized.as_path().ancestors() { + let ancestor = SanitizedPath::unchecked_new(ancestor); + if let Some(registration) = path_registrations.get(ancestor) { + ids.extend_from_slice(®istration.watcher_ids); + } + } + } + ids.sort_unstable_by_key(|id| id.0); + ids.dedup(); + ids.into_iter() + .filter_map(|id| state.watchers.get(&id)) + .map(|registration| registration.callback.clone()) + .collect::>() + } + }; + + for callback in callbacks { + callback(&event); + } + } + fn start_native_watch_limit_cooldown(&self, path: &Path) { let mut state = self.state.lock(); let now = Instant::now(); @@ -671,7 +745,7 @@ impl GlobalWatcher { return; }; drop(state); - self.unwatch(&path, mode).log_err(); + self.unwatch(path.as_path(), mode).log_err(); } fn watch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> { @@ -705,20 +779,28 @@ impl GlobalWatcher { } fn unwatch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> { - match mode { - WatcherMode::Native => { - if let Some(watcher) = self.native_watcher.lock().as_mut() { - watcher.unwatch(path)?; - } - } - WatcherMode::Poll => { - if let Some(watcher) = self.poll_watcher.lock().as_mut() { - watcher.unwatch(path)?; - } - } - } + let watcher = match mode { + WatcherMode::Native => self + .native_watcher + .lock() + .as_mut() + .map(|watcher| watcher.unwatch(path)), + WatcherMode::Poll => self + .poll_watcher + .lock() + .as_mut() + .map(|watcher| watcher.unwatch(path)), + }; - Ok(()) + match watcher { + // inotify auto-removes a watch when its directory is deleted, so a + // later unwatch races that and fails with a benign error. Either way + // the path is no longer watched, which is all we wanted. + Some(Err(error)) if !matches!(error.kind, notify::ErrorKind::WatchNotFound) => { + Err(error.into()) + } + _ => Ok(()), + } } fn ensure_native_watcher(&self) -> anyhow::Result<()> { @@ -732,8 +814,10 @@ impl GlobalWatcher { // risk of queue overflows (and thus full rescans) under read-heavy // workloads like grep or language server indexing. let config = notify::Config::default().with_event_kinds(notify::EventKindMask::CORE); - let watcher = - ::new(handle_native_event, config)?; + let watcher = ::new( + |event| global_watcher().enqueue(WatcherMode::Native, event), + config, + )?; *self.native_watcher.lock() = Some(Box::new(watcher)); Ok(()) } @@ -744,22 +828,26 @@ impl GlobalWatcher { } let config = notify::Config::default().with_poll_interval(*POLL_INTERVAL); - let watcher = notify::PollWatcher::new(handle_poll_event, config)?; + let watcher = notify::PollWatcher::new( + |event| global_watcher().enqueue(WatcherMode::Poll, event), + config, + )?; *self.poll_watcher.lock() = Some(Box::new(watcher)); Ok(()) } } fn path_already_covered( - path: &Path, - path_registrations: &HashMap, PathRegistrationState>, + path: &SanitizedPath, + path_registrations: &HashMap, PathRegistrationState>, mode: WatcherMode, ) -> bool { (mode == WatcherMode::Poll || cfg!(any(target_os = "windows", target_os = "macos"))) && path + .as_path() .ancestors() .skip(1) - .any(|ancestor| path_registrations.contains_key(ancestor)) + .any(|ancestor| path_registrations.contains_key(SanitizedPath::unchecked_new(ancestor))) } fn is_max_files_watch_error(error: &anyhow::Error) -> bool { @@ -793,68 +881,31 @@ pub fn poll_interval() -> Duration { static FS_WATCHER_INSTANCE: OnceLock = OnceLock::new(); fn global_watcher() -> &'static GlobalWatcher { - FS_WATCHER_INSTANCE.get_or_init(|| GlobalWatcher { - state: Mutex::new(WatcherState { - watchers: Default::default(), - native_path_registrations: Default::default(), - poll_path_registrations: Default::default(), - cooldown_until: None, - last_registration: Default::default(), - }), - native_watcher: Mutex::new(None), - poll_watcher: Mutex::new(None), + FS_WATCHER_INSTANCE.get_or_init(|| { + let (event_tx, event_rx) = async_channel::unbounded::(); + 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); + } + }) + .expect("failed to spawn fs watcher dispatch thread"); + GlobalWatcher { + state: Mutex::new(WatcherState { + watchers: Default::default(), + native_path_registrations: Default::default(), + poll_path_registrations: Default::default(), + cooldown_until: None, + last_registration: Default::default(), + }), + native_watcher: Mutex::new(None), + poll_watcher: Mutex::new(None), + event_tx, + } }) } -fn handle_native_event(event: Result) { - handle_event(WatcherMode::Native, event); -} - -fn handle_poll_event(event: Result) { - handle_event(WatcherMode::Poll, event); -} - -fn handle_event(mode: WatcherMode, event: Result) { - if matches!( - event, - Ok(Event { - kind: EventKind::Access(_), - .. - }) - ) { - return; - } - - log::trace!("global handle event for {mode:?}: {event:?}"); - - let callbacks = { - let state = global_watcher().state.lock(); - state - .watchers - .values() - .filter(|registration| registration.mode == mode) - .map(|registration| registration.callback.clone()) - .collect::>() - }; - - match event { - Ok(event) => { - if event.need_rescan() { - log::warn!( - "filesystem watcher lost sync for {mode:?}; scheduling rescans for {} registrations", - callbacks.len() - ); - } - for callback in callbacks { - callback(&event); - } - } - Err(error) => { - log::warn!("watcher error for {mode:?}: {error}"); - } - } -} - #[cfg(test)] mod tests { use super::*; @@ -916,6 +967,10 @@ mod tests { native_watcher: Option>>, poll_watcher: Option>>, ) -> GlobalWatcher { + // Tests call `handle_event` directly to exercise dispatch synchronously, + // rather than going through the OS watcher callbacks, so nothing is ever + // sent on this channel; the receiver can just be dropped. + let (event_tx, _event_rx) = async_channel::unbounded(); GlobalWatcher { state: Mutex::new(WatcherState { watchers: Default::default(), @@ -934,6 +989,7 @@ mod tests { Box::new(SharedFakeWatchBackend(watcher)) as Box }), ), + event_tx, } } @@ -994,6 +1050,111 @@ mod tests { assert_eq!(native_backend.watch_calls, &[first_path.to_path_buf()]); } + fn modify_event(path: &str) -> notify::Event { + notify::Event { + paths: vec![PathBuf::from(path)], + ..notify::Event::new(EventKind::Modify(notify::event::ModifyKind::Any)) + } + } + + fn recording_watcher() -> (GlobalWatcher, Arc>>) { + let backend = Arc::new(Mutex::new(FakeWatchBackend::default())); + let watcher = test_watcher_with_backends(Some(backend), None); + let fired = Arc::new(Mutex::new(Vec::new())); + for dir in ["/repo/a", "/repo/a/nested", "/repo/b"] { + let fired = fired.clone(); + let label = dir.to_owned(); + watcher + .add( + Arc::::from(Path::new(dir)), + WatcherMode::Native, + move |_| { + fired.lock().push(label.clone()); + }, + ) + .expect("add watch") + .expect("watch registered"); + } + (watcher, fired) + } + + #[test] + fn event_dispatches_only_to_registrations_covering_its_path() { + let (watcher, fired) = recording_watcher(); + + watcher.dispatch(WatcherMode::Native, Ok(modify_event("/repo/a/file.txt"))); + + // Only the directory containing the file resolves; siblings stay untouched. + assert_eq!(*fired.lock(), vec!["/repo/a".to_owned()]); + } + + #[test] + fn event_dispatches_to_every_ancestor_registration() { + let (watcher, fired) = recording_watcher(); + + watcher.dispatch( + WatcherMode::Native, + Ok(modify_event("/repo/a/nested/file.txt")), + ); + + // Both the directory containing the file and the ancestor watching it + // resync, each exactly once, matching the previous broadcast-and-filter + // behavior. + let mut got = fired.lock().clone(); + got.sort(); + assert_eq!(got, vec!["/repo/a".to_owned(), "/repo/a/nested".to_owned()]); + } + + #[test] + fn rescan_event_broadcasts_to_all_registrations_of_the_same_mode() { + let (watcher, fired) = recording_watcher(); + + let rescan = notify::Event::new(EventKind::Other).set_flag(notify::event::Flag::Rescan); + watcher.dispatch(WatcherMode::Native, Ok(rescan)); + + // A pathless rescan may have missed events anywhere, so every registration + // of that mode resyncs, regardless of which directory it watches. + let mut got = fired.lock().clone(); + got.sort(); + assert_eq!( + got, + vec![ + "/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() { + // `notify` (and Windows APIs generally) can report a changed path using the + // verbatim `\\?\` long-path form even when the directory was registered + // without it. + let backend = Arc::new(Mutex::new(FakeWatchBackend::default())); + let watcher = test_watcher_with_backends(Some(backend), None); + let fired = Arc::new(Mutex::new(Vec::new())); + { + let fired = fired.clone(); + watcher + .add( + Arc::::from(Path::new("C:\\repo\\src")), + WatcherMode::Native, + move |_| fired.lock().push(()), + ) + .expect("add watch") + .expect("watch registered"); + } + + watcher.dispatch( + WatcherMode::Native, + Ok(modify_event("\\\\?\\C:\\repo\\src\\main.rs")), + ); + + assert_eq!(fired.lock().len(), 1); + } + #[test] fn test_coalesce_pending_rescans() { let test_cases = [