From 102a6c5d8a095ee50fd0481d30caf488a6263572 Mon Sep 17 00:00:00 2001 From: Cole Miller Date: Mon, 7 Sep 2026 19:01:41 +0000 Subject: [PATCH] fs: Refactor the code around `GlobalWatcher` to be more testable (#63894) This is a pure refactoring PR to make our FS watching code more testable. It does two things: 1. Make the `GlobalWatcher` be owned by `RealFs` or `FakeFs` instead of being a process-wide global. 2. Delete `FakeWatcher` and its `Watcher` implementation, and instead mock the notify crate itself in `FakeFs` tests via the existing `WatchBackend` trait. Now all our FS watching code down to the notify boundary runs the same in tests and production. Release Notes: - N/A --- crates/agent/src/agent.rs | 2 +- .../examples/component_preview.rs | 2 +- crates/edit_prediction_cli/src/headless.rs | 5 +- crates/eval_cli/src/headless.rs | 5 +- crates/extension_cli/src/main.rs | 2 +- .../extension_compilation_benchmark.rs | 8 +- .../src/extension_store_test.rs | 2 +- crates/fs/src/fs.rs | 393 +++++++-------- crates/fs/src/fs_watcher.rs | 472 ++++++++++++------ crates/project/src/agent_server_store.rs | 2 +- crates/project/src/project.rs | 2 +- .../tests/integration/project_tests.rs | 49 +- crates/project_benchmarks/src/main.rs | 2 +- crates/remote_server/src/server.rs | 2 +- crates/repl/src/repl_editor.rs | 2 +- crates/vim/src/test.rs | 2 +- crates/workspace/src/multi_workspace.rs | 9 +- crates/worktree/src/worktree.rs | 2 +- .../tests/integration/worktree_tests.rs | 50 +- crates/worktree_benchmarks/src/main.rs | 2 +- crates/zed/src/main.rs | 2 +- crates/zed/src/visual_test_runner.rs | 2 +- crates/zed/src/zed.rs | 11 +- 23 files changed, 582 insertions(+), 448 deletions(-) diff --git a/crates/agent/src/agent.rs b/crates/agent/src/agent.rs index 252a011d47..998c0e16f1 100644 --- a/crates/agent/src/agent.rs +++ b/crates/agent/src/agent.rs @@ -3898,7 +3898,7 @@ mod internal_tests { }); let temp_dir = tempfile::tempdir().expect("create terminal working directory"); - let fs = Arc::new(fs::RealFs::new(None, cx.executor())); + let fs = fs::RealFs::new(None, cx.executor()); let project = Project::test(fs.clone(), [temp_dir.path()], cx).await; let thread_store = cx.new(|cx| ThreadStore::new(cx)); let agent = cx.update(|cx| NativeAgent::new(thread_store, Templates::new(), fs, cx)); diff --git a/crates/component_preview/examples/component_preview.rs b/crates/component_preview/examples/component_preview.rs index a65821d071..0af577b6b4 100644 --- a/crates/component_preview/examples/component_preview.rs +++ b/crates/component_preview/examples/component_preview.rs @@ -44,7 +44,7 @@ fn main() { ReqwestClient::user_agent("component_preview").expect("Failed to create HTTP client"); cx.set_http_client(Arc::new(http_client)); - let fs = Arc::new(RealFs::new(None, cx.background_executor().clone())); + let fs = RealFs::new(None, cx.background_executor().clone()); ::set_global(fs.clone(), cx); settings::init(cx); diff --git a/crates/edit_prediction_cli/src/headless.rs b/crates/edit_prediction_cli/src/headless.rs index 48b7381020..b19c2612b2 100644 --- a/crates/edit_prediction_cli/src/headless.rs +++ b/crates/edit_prediction_cli/src/headless.rs @@ -66,10 +66,7 @@ pub fn init(cx: &mut App) -> EpAppState { cx.set_global(app_db); let git_binary_path = None; - let fs = Arc::new(RealFs::new( - git_binary_path, - cx.background_executor().clone(), - )); + let fs = RealFs::new(git_binary_path, cx.background_executor().clone()); let mut languages = LanguageRegistry::new(cx.background_executor().clone()); languages.set_language_server_download_dir(paths::languages_dir().clone()); diff --git a/crates/eval_cli/src/headless.rs b/crates/eval_cli/src/headless.rs index 156c549ac8..7739dcc6e0 100644 --- a/crates/eval_cli/src/headless.rs +++ b/crates/eval_cli/src/headless.rs @@ -67,10 +67,7 @@ pub fn init(cx: &mut App) -> Arc { cx.set_global(app_db); let git_binary_path = None; - let fs = Arc::new(RealFs::new( - git_binary_path, - cx.background_executor().clone(), - )); + let fs = RealFs::new(git_binary_path, cx.background_executor().clone()); ::set_global(fs.clone(), cx); let mut languages = LanguageRegistry::new(cx.background_executor().clone()); diff --git a/crates/extension_cli/src/main.rs b/crates/extension_cli/src/main.rs index f31a400467..d9c823e370 100644 --- a/crates/extension_cli/src/main.rs +++ b/crates/extension_cli/src/main.rs @@ -66,7 +66,7 @@ async fn main() -> Result<()> { env_logger::init(); let args = Args::parse(); - let fs = Arc::new(RealFs::new(None, gpui_platform::background_executor())); + let fs = RealFs::new(None, gpui_platform::background_executor()); let engine = wasmtime::Engine::default(); let mut wasm_store = WasmStore::new(&engine)?; diff --git a/crates/extension_host/benches/extension_compilation_benchmark.rs b/crates/extension_host/benches/extension_compilation_benchmark.rs index e8abdffbd8..73fe3fd246 100644 --- a/crates/extension_host/benches/extension_compilation_benchmark.rs +++ b/crates/extension_host/benches/extension_compilation_benchmark.rs @@ -24,11 +24,7 @@ fn extension_benchmarks(c: &mut Criterion) { let mut group = c.benchmark_group("load"); let mut manifest = manifest(); - let wasm_bytes = wasm_bytes( - &cx, - &mut manifest, - Arc::new(RealFs::new(None, cx.executor())), - ); + let wasm_bytes = wasm_bytes(&cx, &mut manifest, RealFs::new(None, cx.executor())); let manifest = Arc::new(manifest); let extensions_dir = TempTree::new(json!({ "installed": {}, @@ -105,7 +101,7 @@ fn wasm_host(cx: &TestAppContext, extensions_dir: &TempTree) -> Arc { }); let extensions_dir = extensions_dir.path().canonicalize().unwrap(); let work_dir = extensions_dir.join("work"); - let fs = Arc::new(RealFs::new(None, cx.executor())); + let fs = RealFs::new(None, cx.executor()); cx.update(|cx| { WasmHost::new( diff --git a/crates/extension_host/src/extension_store_test.rs b/crates/extension_host/src/extension_store_test.rs index ad6679e8fe..6fe0650788 100644 --- a/crates/extension_host/src/extension_store_test.rs +++ b/crates/extension_host/src/extension_store_test.rs @@ -802,7 +802,7 @@ async fn test_extension_store_with_test_extension(cx: &mut TestAppContext) { let test_extension_id = "test-extension"; let test_extension_dir = root_dir.join("extensions").join(test_extension_id); - let fs = Arc::new(RealFs::new(None, cx.executor())); + let fs = RealFs::new(None, cx.executor()); let extensions_tree = TempTree::new(json!({ "installed": {}, "work": {} diff --git a/crates/fs/src/fs.rs b/crates/fs/src/fs.rs index e775b85534..ad9ba139a9 100644 --- a/crates/fs/src/fs.rs +++ b/crates/fs/src/fs.rs @@ -1,8 +1,6 @@ pub mod fs_watcher; mod git_clone_progress; -pub use fs_watcher::requires_poll_watcher; - use parking_lot::Mutex; use slotmap::{KeyData, SlotMap}; use std::ffi::OsString; @@ -148,6 +146,21 @@ pub trait Fs: Send + Sync { path: &Path, ) -> Result>>>>; + /// Creates the native file watcher now rather than on the first `watch`, so a + /// failure to start it (e.g. inotify instance limits) can be reported at startup. + fn start_native_watcher(&self) -> Result<()> { + Ok(()) + } + + /// Whether `path` exists, without following a final symlink. Synchronous + /// because watches are registered synchronously by the worktree scanner. + fn path_exists(&self, path: &Path) -> bool; + /// Whether the volume holding `path` compares file names case-sensitively. + fn is_path_case_sensitive(&self, path: &Path) -> bool; + /// Whether `path` sits on a filesystem where native file watching does not + /// deliver events (network mounts, some FUSE and WSL mounts), so it must be polled. + fn requires_poll_watcher(&self, path: &Path) -> bool; + async fn watch( &self, path: &Path, @@ -430,8 +443,10 @@ impl TrashId { } pub struct RealFs { + this: std::sync::Weak, bundled_git_binary_path: Option, executor: BackgroundExecutor, + global_watcher: Arc, next_job_id: Arc, job_event_subscribers: Arc>>, trash: Arc>>, @@ -537,15 +552,17 @@ impl FileHandle for std::fs::File { pub struct RealWatcher {} impl RealFs { - pub fn new(git_binary_path: Option, executor: BackgroundExecutor) -> Self { - Self { + pub fn new(git_binary_path: Option, executor: BackgroundExecutor) -> Arc { + Arc::new_cyclic(|this| Self { + this: this.clone(), bundled_git_binary_path: git_binary_path, + global_watcher: fs_watcher::GlobalWatcher::new(executor.clone()), executor, next_job_id: Arc::new(AtomicUsize::new(0)), job_event_subscribers: Arc::new(Mutex::new(Vec::new())), trash: Arc::new(Mutex::new(SlotMap::with_key())), is_case_sensitive: Default::default(), - } + }) } #[cfg(target_os = "windows")] @@ -1146,6 +1163,22 @@ impl Fs for RealFs { Ok(Box::pin(iter(entries))) } + fn start_native_watcher(&self) -> Result<()> { + self.global_watcher.ensure_native_watcher() + } + + fn path_exists(&self, path: &Path) -> bool { + std::fs::symlink_metadata(path).is_ok() + } + + fn is_path_case_sensitive(&self, path: &Path) -> bool { + !fs_watcher::case_insensitive_path(path) + } + + fn requires_poll_watcher(&self, path: &Path) -> bool { + fs_watcher::requires_poll_watcher(path) + } + async fn watch( &self, path: &Path, @@ -1154,62 +1187,18 @@ impl Fs for RealFs { Pin>>>, Arc, ) { - use util::{ResultExt as _, paths::SanitizedPath}; - let executor = self.executor.clone(); - - let (tx, rx) = async_channel::unbounded(); - let pending_paths: Arc>> = Default::default(); - - let watcher: Arc = Arc::new(fs_watcher::FsWatcher::new( - executor.clone(), - tx.clone(), - pending_paths.clone(), - )); - - if let Err(e) = watcher.add(path) { - log::warn!("Failed to watch {}:\n{e}", path.display()); - } - - // Check if path is a symlink and follow the target parent - if let Some(mut target) = self.read_link(path).await.ok() { - log::trace!("watch symlink {path:?} -> {target:?}"); - // Check if symlink target is relative path, if so make it absolute - if target.is_relative() - && let Some(parent) = path.parent() - { - target = parent.join(target); - if let Ok(canonical) = self.canonicalize(&target).await { - target = SanitizedPath::new(&canonical).as_path().to_path_buf(); - } - } - watcher.add(&target).ok(); - // Skipped for poll watchers: PollWatcher::watch() recursively scans - // at registration, blocking on large virtual filesystem mounts - if let Some(parent) = target.parent() - && !fs_watcher::requires_poll_watcher(parent) - { - watcher.add(parent).log_err(); - } - } - - ( - Box::pin(rx.filter_map({ - let watcher = watcher.clone(); - let executor = executor.clone(); - move |_| { - let _ = watcher.clone(); - let pending_paths = pending_paths.clone(); - let executor = executor.clone(); - async move { - executor.timer(latency).await; - let paths = std::mem::take(&mut *pending_paths.lock()); - log::debug!("pending path events: {:?}", paths); - (!paths.is_empty()).then_some(paths) - } - } - })), - watcher, + let this = self + .this + .upgrade() + .expect("RealFs is only constructed inside an Arc"); + fs_watcher::watch( + this, + self.global_watcher.clone(), + self.executor.clone(), + path, + latency, ) + .await } fn open_repo( @@ -1430,6 +1419,7 @@ pub struct FakeFs { // Use an unfair lock to ensure tests are deterministic. state: Arc>, executor: gpui::BackgroundExecutor, + global_watcher: Arc, } #[cfg(feature = "test-support")] @@ -1438,7 +1428,8 @@ struct FakeFsState { next_inode: u64, next_mtime: SystemTime, git_event_tx: async_channel::Sender, - event_txs: Vec<(PathBuf, async_channel::Sender>)>, + watch_roots: Vec<(PathBuf, std::sync::Weak)>, + watches: FakeWatches, events_paused: bool, buffered_events: Vec, metadata_call_count: usize, @@ -1446,41 +1437,19 @@ struct FakeFsState { path_write_counts: std::collections::HashMap, job_event_subscribers: Arc>>, trash: Mutex>, - file_to_create_before_watch_add: Option<(PathBuf, PathBuf)>, remove_dir_errors: std::collections::HashMap, case_sensitive: bool, } +/// The kernel's side of file watching, as far as the real watcher code above +/// notify can tell: the paths the backend has registered and the callback that +/// delivers events for them into the `GlobalWatcher`. #[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(()) - } +#[derive(Default)] +struct FakeWatches { + registered_paths: Vec, + watch_calls: Vec, + event_sink: Option) + Send + Sync>>, } #[cfg(feature = "test-support")] @@ -1735,12 +1704,73 @@ impl FakeFsState { } fn flush_events(&mut self, mut count: usize) { + use notify::event::{CreateKind, Flag, ModifyKind, RemoveKind}; + count = count.min(self.buffered_events.len()); let events = self.buffered_events.drain(0..count).collect::>(); - self.event_txs.retain(|(_, tx)| { - let _ = tx.try_send(events.clone()); - !tx.is_closed() - }); + let Some(event_sink) = &self.watches.event_sink else { + return; + }; + for event in events { + let is_registered = self.watches.registered_paths.iter().any(|registered_path| { + if self.case_sensitive { + event.path.starts_with(registered_path) + } else { + let event_path = event.path.to_string_lossy().to_lowercase(); + let registered_path = registered_path.to_string_lossy().to_lowercase(); + Path::new(&event_path).starts_with(Path::new(®istered_path)) + } + }); + if !is_registered { + continue; + } + let notify_event = match event.kind { + Some(PathEventKind::Created) => { + notify::Event::new(notify::EventKind::Create(CreateKind::Any)) + } + Some(PathEventKind::Changed) => { + notify::Event::new(notify::EventKind::Modify(ModifyKind::Any)) + } + Some(PathEventKind::Removed) => { + notify::Event::new(notify::EventKind::Remove(RemoveKind::Any)) + } + Some(PathEventKind::Rescan) => { + notify::Event::new(notify::EventKind::Other).set_flag(Flag::Rescan) + } + None => notify::Event::new(notify::EventKind::Any), + }; + event_sink(Ok(notify_event.add_path(event.path))); + } + } +} + +/// Stands in for notify at the boundary the real watcher code talks to: the +/// fake filesystem's own mutations are the "kernel" events. +#[cfg(feature = "test-support")] +struct FakeWatchBackend { + state: Arc>, +} + +#[cfg(feature = "test-support")] +impl fs_watcher::WatchBackend for FakeWatchBackend { + fn watch(&mut self, path: &Path, _mode: notify::RecursiveMode) -> notify::Result<()> { + let path = normalize_path(path); + let mut state = self.state.try_lock().expect( + "fake filesystem state is locked; this execution would have caused a test hang", + ); + state.watches.watch_calls.push(path.clone()); + state.watches.registered_paths.push(path); + Ok(()) + } + + fn unwatch(&mut self, path: &Path) -> notify::Result<()> { + let path = normalize_path(path); + self.state + .lock() + .watches + .registered_paths + .retain(|registered_path| *registered_path != path); + Ok(()) } } @@ -1757,32 +1787,42 @@ impl FakeFs { pub fn new(executor: gpui::BackgroundExecutor) -> Arc { let (tx, rx) = async_channel::bounded::(10); + let state = Arc::new(Mutex::new(FakeFsState { + root: FakeFsEntry::Dir { + inode: 0, + mtime: MTime(UNIX_EPOCH), + len: 0, + entries: Default::default(), + git_repo_state: None, + }, + git_event_tx: tx, + next_mtime: UNIX_EPOCH + Self::SYSTEMTIME_INTERVAL, + next_inode: 1, + watch_roots: Vec::new(), + watches: FakeWatches::default(), + buffered_events: Vec::new(), + events_paused: false, + read_dir_call_count: 0, + metadata_call_count: 0, + path_write_counts: Default::default(), + job_event_subscribers: Arc::new(Mutex::new(Vec::new())), + trash: Mutex::new(SlotMap::with_key()), + remove_dir_errors: Default::default(), + case_sensitive: true, + })); + let global_watcher = fs_watcher::GlobalWatcher::with_native_backend( + executor.clone(), + Some(Box::new(FakeWatchBackend { + state: state.clone(), + })), + ); + state.lock().watches.event_sink = Some(Box::new(global_watcher.native_event_sink())); + let this = Arc::new_cyclic(|this| Self { this: this.clone(), executor: executor.clone(), - state: Arc::new(Mutex::new(FakeFsState { - root: FakeFsEntry::Dir { - inode: 0, - mtime: MTime(UNIX_EPOCH), - len: 0, - entries: Default::default(), - git_repo_state: None, - }, - git_event_tx: tx, - next_mtime: UNIX_EPOCH + Self::SYSTEMTIME_INTERVAL, - next_inode: 1, - event_txs: Default::default(), - buffered_events: Vec::new(), - events_paused: false, - read_dir_call_count: 0, - metadata_call_count: 0, - path_write_counts: Default::default(), - job_event_subscribers: Arc::new(Mutex::new(Vec::new())), - trash: Mutex::new(SlotMap::with_key()), - file_to_create_before_watch_add: None, - remove_dir_errors: Default::default(), - case_sensitive: true, - })), + state, + global_watcher, }); executor.spawn({ @@ -1976,15 +2016,10 @@ impl FakeFs { state.emit_event([(root, Some(PathEventKind::Rescan))]); } - pub fn create_file_before_next_watch_add( - &self, - watch_path: impl AsRef, - path: impl AsRef, - ) { - self.state.lock().file_to_create_before_watch_add = Some(( - normalize_path(watch_path.as_ref()), - normalize_path(path.as_ref()), - )); + /// Every path the watcher backend has been asked to watch, in order, + /// including paths that were later unwatched. + pub fn watch_calls(&self) -> Vec { + self.state.lock().watches.watch_calls.clone() } pub fn flush_events(&self, count: usize) { @@ -2664,12 +2699,13 @@ impl FakeFs { self.state.lock().read_dir_call_count } + /// The roots passed to `Fs::watch` whose watchers are still alive. pub fn watched_paths(&self) -> Vec { let state = self.state.lock(); state - .event_txs + .watch_roots .iter() - .filter_map(|(path, tx)| (!tx.is_closed()).then_some(path.clone())) + .filter_map(|(path, watcher)| (watcher.strong_count() > 0).then_some(path.clone())) .collect() } @@ -2821,48 +2857,6 @@ impl FakeFsEntry { } } -#[cfg(feature = "test-support")] -struct FakeWatcher { - tx: async_channel::Sender>, - fs_state: Arc>, - prefixes: Mutex>, -} - -#[cfg(feature = "test-support")] -impl Watcher for FakeWatcher { - fn add(&self, path: &Path) -> Result<()> { - 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.clone(), self.tx.clone())); - prefixes.push(path); - Ok(()) - } - - 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(()) - } -} - #[cfg(feature = "test-support")] #[derive(Debug)] struct FakeHandle { @@ -3333,35 +3327,21 @@ impl Fs for FakeFs { Arc, ) { self.simulate_random_delay().await; - let (tx, rx) = async_channel::unbounded(); - let path = path.to_path_buf(); - self.state.lock().event_txs.push((path.clone(), tx.clone())); - let executor = self.executor.clone(); - let watcher = Arc::new(FakeWatcher { - tx, - fs_state: self.state.clone(), - prefixes: Mutex::new(vec![path]), - }); - ( - Box::pin(futures::StreamExt::filter(rx, { - let watcher = watcher.clone(); - move |events| { - let result = events.iter().any(|evt_path| { - watcher - .prefixes - .lock() - .iter() - .any(|prefix| evt_path.path.starts_with(prefix)) - }); - let executor = executor.clone(); - async move { - executor.simulate_random_delay().await; - result - } - } - })), - watcher, + // Zero latency: the deterministic executor doesn't advance time on its + // own, so a real debounce would stall every test until `advance_clock`. + let (events, watcher) = fs_watcher::watch( + self.this.upgrade().unwrap(), + self.global_watcher.clone(), + self.executor.clone(), + path, + Duration::ZERO, ) + .await; + self.state + .lock() + .watch_roots + .push((normalize_path(path), Arc::downgrade(&watcher))); + (events, watcher) } fn open_repo( @@ -3402,6 +3382,21 @@ impl Fs for FakeFs { anyhow::bail!("Git config is not supported in fake Fs") } + fn path_exists(&self, path: &Path) -> bool { + self.state + .lock() + .try_entry(&normalize_path(path), false) + .is_some() + } + + fn is_path_case_sensitive(&self, _path: &Path) -> bool { + self.state.lock().case_sensitive + } + + fn requires_poll_watcher(&self, _path: &Path) -> bool { + false + } + fn is_fake(&self) -> bool { true } diff --git a/crates/fs/src/fs_watcher.rs b/crates/fs/src/fs_watcher.rs index 59ee6ac984..431723fca3 100644 --- a/crates/fs/src/fs_watcher.rs +++ b/crates/fs/src/fs_watcher.rs @@ -1,17 +1,18 @@ +use futures::{Stream, StreamExt as _}; use gpui::{BackgroundExecutor, Task}; use notify::{Event, EventKind}; use parking_lot::Mutex; use std::{ collections::HashMap, - fs, ops::DerefMut, path::Path, - sync::{Arc, LazyLock, OnceLock}, + pin::Pin, + sync::{Arc, LazyLock}, time::{Duration, Instant}, }; use util::{ResultExt, paths::SanitizedPath}; -use crate::{PathEvent, PathEventKind, Watcher}; +use crate::{Fs, PathEvent, PathEventKind, Watcher}; #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] pub enum WatcherMode { @@ -20,7 +21,77 @@ pub enum WatcherMode { Poll, } +pub(crate) async fn watch( + fs: Arc, + global_watcher: Arc, + executor: BackgroundExecutor, + path: &Path, + latency: Duration, +) -> ( + Pin>>>, + Arc, +) { + let (tx, rx) = async_channel::unbounded(); + let pending_paths: Arc>> = Default::default(); + + let watcher: Arc = Arc::new(FsWatcher::new( + global_watcher, + fs.clone(), + executor.clone(), + tx, + pending_paths.clone(), + )); + + if let Err(e) = watcher.add(path) { + log::warn!("Failed to watch {}:\n{e}", path.display()); + } + + // Check if path is a symlink and follow the target parent + if let Some(mut target) = fs.read_link(path).await.ok() { + log::trace!("watch symlink {path:?} -> {target:?}"); + // Check if symlink target is relative path, if so make it absolute + if target.is_relative() + && let Some(parent) = path.parent() + { + target = parent.join(target); + if let Ok(canonical) = fs.canonicalize(&target).await { + target = SanitizedPath::new(&canonical).as_path().to_path_buf(); + } + } + watcher.add(&target).ok(); + // Skipped for poll watchers: PollWatcher::watch() recursively scans + // at registration, blocking on large virtual filesystem mounts + if let Some(parent) = target.parent() + && !fs.requires_poll_watcher(parent) + { + watcher.add(parent).log_err(); + } + } + + ( + Box::pin(rx.filter_map({ + let watcher = watcher.clone(); + move |_| { + let _ = watcher.clone(); + let pending_paths = pending_paths.clone(); + let executor = executor.clone(); + async move { + if !latency.is_zero() { + executor.timer(latency).await; + } + let paths = std::mem::take(&mut *pending_paths.lock()); + log::debug!("pending path events: {:?}", paths); + (!paths.is_empty()).then_some(paths) + } + } + })), + watcher, + ) +} + pub struct FsWatcher { + global_watcher: Arc, + fs: Arc, executor: BackgroundExecutor, tx: async_channel::Sender<()>, pending_path_events: Arc>>, @@ -35,12 +106,16 @@ struct FsWatcherRegistration { } impl FsWatcher { - pub fn new( + pub(crate) fn new( + global_watcher: Arc, + fs: Arc, executor: BackgroundExecutor, tx: async_channel::Sender<()>, pending_path_events: Arc>>, ) -> Self { Self { + global_watcher, + fs, executor, tx, pending_path_events, @@ -50,13 +125,15 @@ impl FsWatcher { } fn add_existing_path(&self, path: Arc) -> anyhow::Result<()> { - let case_insensitive = case_insensitive_path(&path); + let case_insensitive = !self.fs.is_path_case_sensitive(&path); let key = WatchKey::for_registration(SanitizedPath::new(&path), case_insensitive); if self.registrations.lock().contains_key(&key) { log::trace!("path to watch is already watched: {path:?}"); return Ok(()); } match register_existing_path( + &self.global_watcher, + self.fs.as_ref(), path.clone(), case_insensitive, self.tx.clone(), @@ -83,6 +160,8 @@ impl FsWatcher { } let task = self.executor.spawn(poll_path_until_created( + self.global_watcher.clone(), + self.fs.clone(), self.executor.clone(), path.clone(), self.tx.clone(), @@ -104,9 +183,8 @@ impl Drop for FsWatcher { std::mem::swap(old.deref_mut(), &mut registrations); } - let global_watcher = global_watcher(); for (_, registration) in registrations { - global_watcher.remove(registration.id); + self.global_watcher.remove(registration.id); } } } @@ -133,7 +211,7 @@ impl Watcher for FsWatcher { return Ok(()); } - if fs::symlink_metadata(path.as_ref()).is_err() { + if !self.fs.path_exists(&path) { self.add_pending_path(path); return Ok(()); } @@ -153,7 +231,7 @@ impl Watcher for FsWatcher { .or_else(|| registrations.remove(&WatchKey::folded(sanitized))) }; if let Some(registration) = registration { - global_watcher().remove(registration.id); + self.global_watcher.remove(registration.id); } Ok(()) } @@ -211,12 +289,14 @@ pub fn requires_poll_watcher(path: &Path) -> bool { } fn register_existing_path( + global_watcher: &GlobalWatcher, + fs: &dyn Fs, path: Arc, case_insensitive: bool, tx: async_channel::Sender<()>, pending_path_events: Arc>>, ) -> anyhow::Result> { - let mode = if requires_poll_watcher(path.as_ref()) { + let mode = if fs.requires_poll_watcher(path.as_ref()) { log::info!( "Using poll watcher ({}ms interval) for {}", poll_interval().as_millis(), @@ -229,7 +309,7 @@ fn register_existing_path( }; let root_path = SanitizedPath::new_arc(path.as_ref()); let path_for_callback = path.clone(); - let Some(registration_id) = global_watcher().add( + let Some(registration_id) = global_watcher.add( path, mode, case_insensitive, @@ -365,7 +445,7 @@ fn is_wsl_drvfs_path(path: &Path) -> bool { /// Whether the volume backing `path` does case-insensitive name lookups, used to /// pick exact vs. folded matching. #[cfg(target_os = "macos")] -fn case_insensitive_path(path: &Path) -> bool { +pub(crate) fn case_insensitive_path(path: &Path) -> bool { use std::os::unix::ffi::OsStrExt as _; // `pathconf(_PC_CASE_SENSITIVE)` returns 1 (sensitive), 0 (insensitive), or -1 @@ -378,7 +458,7 @@ fn case_insensitive_path(path: &Path) -> bool { } #[cfg(target_os = "linux")] -fn case_insensitive_path(_path: &Path) -> bool { +pub(crate) fn case_insensitive_path(_path: &Path) -> bool { // use std::os::unix::ffi::OsStrExt as _; // // Only ext4/f2fs casefold (`+F`) dirs are insensitive, reported by `statx` via @@ -403,7 +483,7 @@ fn case_insensitive_path(_path: &Path) -> bool { } #[cfg(target_os = "windows")] -fn case_insensitive_path(_path: &Path) -> bool { +pub(crate) fn case_insensitive_path(_path: &Path) -> bool { // todo(windows): Windows defaults to case in sensitive, but // they can mark specific directories as case sensitive. Mainly // for WSL use cases @@ -411,7 +491,7 @@ fn case_insensitive_path(_path: &Path) -> bool { } #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))] -fn case_insensitive_path(_path: &Path) -> bool { +pub(crate) fn case_insensitive_path(_path: &Path) -> bool { // Other BSDs default to case-sensitive local filesystems. false } @@ -480,6 +560,8 @@ impl WatchKey { } async fn poll_path_until_created( + global_watcher: Arc, + fs: Arc, executor: BackgroundExecutor, path: Arc, tx: async_channel::Sender<()>, @@ -494,13 +576,13 @@ async fn poll_path_until_created( return; } - if smol::fs::symlink_metadata(path.as_ref()).await.is_err() { + if !fs.path_exists(&path) { continue; } // Probe case sensitivity now that the path exists, rather than at add // time when it didn't. - let case_insensitive = case_insensitive_path(path.as_ref()); + let case_insensitive = !fs.is_path_case_sensitive(&path); let key = WatchKey::for_registration(SanitizedPath::new(&path), case_insensitive); if registrations.lock().contains_key(&key) { @@ -509,6 +591,8 @@ async fn poll_path_until_created( } match register_existing_path( + &global_watcher, + fs.as_ref(), path.clone(), case_insensitive, tx.clone(), @@ -518,7 +602,7 @@ async fn poll_path_until_created( { let mut pending_registrations = pending_registrations.lock(); if pending_registrations.remove(path.as_ref()).is_none() { - global_watcher().remove(registration.id); + global_watcher.remove(registration.id); return; } registrations.lock().insert(key, registration); @@ -598,11 +682,19 @@ fn push_notify_event( log::warn!("filesystem watcher lost sync for {watched_root:?}; scheduling rescan"); } - path_events.retain(|path_event| path_event.path != watched_root); - path_events.push(PathEvent { - path: watched_root.to_path_buf(), - kind: Some(PathEventKind::Rescan), - }); + // A rescan event names the subtrees that lost sync (FSEvents + // `MustScanSubDirs`); only a pathless one (inotify queue overflow) means + // the whole watched root is suspect. + if path_events.is_empty() { + path_events.push(PathEvent { + path: watched_root.to_path_buf(), + kind: Some(PathEventKind::Rescan), + }); + } else { + for path_event in &mut path_events { + path_event.kind = Some(PathEventKind::Rescan); + } + } } log::trace!("path_events: {:?}", path_events); enqueue_path_events(tx, pending_path_events, path_events); @@ -789,7 +881,7 @@ impl WatcherState { } } -trait WatchBackend: Send { +pub(crate) trait WatchBackend: Send { fn watch(&mut self, path: &Path, mode: notify::RecursiveMode) -> notify::Result<()>; fn unwatch(&mut self, path: &Path) -> notify::Result<()>; } @@ -807,16 +899,63 @@ impl WatchBackend for T { type DispatchEvent = (WatcherMode, Result); pub struct GlobalWatcher { - state: Mutex, + state: Arc>, // DANGER: never keep state lock while holding watcher lock // two mutexes because calling watcher.add triggers watcher.event, which needs watchers. native_watcher: Mutex>>, poll_watcher: Mutex>>, event_tx: async_channel::Sender, + _dispatch_task: Task<()>, } impl GlobalWatcher { + /// Watches with notify's OS backends, created lazily on first use so that a + /// failure to start them can be reported by `ensure_native_watcher`. + pub fn new(executor: BackgroundExecutor) -> Arc { + Self::with_native_backend(executor, None) + } + + /// Watches with the given backend in place of the OS one; the returned sink + /// is how that backend reports events, standing in for notify's `EventHandler`. + pub(crate) fn with_native_backend( + executor: BackgroundExecutor, + native_watcher: Option>, + ) -> Arc { + let (event_tx, event_rx) = async_channel::unbounded::(); + let state = Arc::new(Mutex::new(WatcherState { + watchers: Default::default(), + native_path_registrations: Default::default(), + poll_path_registrations: Default::default(), + cooldown_until: None, + last_registration: Default::default(), + })); + let dispatch_task = executor.spawn({ + let state = state.clone(); + async move { + while let Ok(first) = event_rx.recv().await { + dispatch_batch(&state, first, &event_rx); + } + } + }); + Arc::new(Self { + state, + native_watcher: Mutex::new(native_watcher), + poll_watcher: Mutex::new(None), + event_tx, + _dispatch_task: dispatch_task, + }) + } + + /// The callback a substitute native backend uses to report events, standing + /// in for the `EventHandler` notify would have been constructed with. + pub(crate) fn native_event_sink( + &self, + ) -> impl Fn(notify::Result) + Send + Sync + 'static { + let event_tx = self.event_tx.clone(); + move |event| enqueue(&event_tx, WatcherMode::Native, event) + } + #[must_use] fn add( &self, @@ -875,97 +1014,18 @@ impl GlobalWatcher { 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(); - } - + #[cfg(test)] 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); - path_registrations.watcher_ids_covering(sanitized, &mut 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); - } + dispatch(&self.state, mode, event); } + #[cfg(test)] fn dispatch_batch( &self, first: DispatchEvent, event_rx: &async_channel::Receiver, ) { - // 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); - } + dispatch_batch(&self.state, first, event_rx); } fn start_native_watch_limit_cooldown(&self, path: &Path) { @@ -1045,7 +1105,7 @@ impl GlobalWatcher { } } - fn ensure_native_watcher(&self) -> anyhow::Result<()> { + pub(crate) fn ensure_native_watcher(&self) -> anyhow::Result<()> { // The lock is held across creation: with a check-then-insert under two // separate lock acquisitions, concurrent callers could each create a // watcher and the loser's insert would silently drop the winner's @@ -1059,7 +1119,7 @@ impl GlobalWatcher { // workloads like grep or language server indexing. let config = notify::Config::default().with_event_kinds(notify::EventKindMask::CORE); let watcher = ::new( - |event| global_watcher().enqueue(WatcherMode::Native, event), + self.native_event_sink(), config, )?; *native_watcher = Some(Box::new(watcher)); @@ -1071,8 +1131,9 @@ impl GlobalWatcher { let mut poll_watcher = self.poll_watcher.lock(); if poll_watcher.is_none() { let config = notify::Config::default().with_poll_interval(*POLL_INTERVAL); + let event_tx = self.event_tx.clone(); let watcher = notify::PollWatcher::new( - |event| global_watcher().enqueue(WatcherMode::Poll, event), + move |event| enqueue(&event_tx, WatcherMode::Poll, event), config, )?; *poll_watcher = Some(Box::new(watcher)); @@ -1081,6 +1142,112 @@ impl GlobalWatcher { } } +fn dispatch( + state: &Mutex, + 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 = 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); + path_registrations.watcher_ids_covering(sanitized, &mut 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 dispatch_batch( + state: &Mutex, + first: DispatchEvent, + event_rx: &async_channel::Receiver, +) { + // A single backend overflow can enqueue many pathless rescan markers, each + // of which rescans every watched root, so one per mode covers the entire + // drained batch. Rescans that name paths only cover those subtrees and are + // all dispatched; 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(|event| event.need_rescan() && event.paths.is_empty()) + { + if *rescan_dispatched { + continue; + } + *rescan_dispatched = true; + } + + dispatch(state, mode, event); + } +} + +fn enqueue( + event_tx: &async_channel::Sender, + 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. + event_tx.try_send((mode, event)).ok(); +} + fn is_max_files_watch_error(error: &anyhow::Error) -> bool { error .downcast_ref::() @@ -1109,34 +1276,6 @@ pub fn poll_interval() -> Duration { *POLL_INTERVAL } -static FS_WATCHER_INSTANCE: OnceLock = OnceLock::new(); - -fn global_watcher() -> &'static GlobalWatcher { - 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(first) = event_rx.recv_blocking() { - global_watcher().dispatch_batch(first, &event_rx); - } - }) - .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, - } - }) -} - #[cfg(test)] mod tests { use super::*; @@ -1203,13 +1342,13 @@ mod tests { // sent on this channel; the receiver can just be dropped. let (event_tx, _event_rx) = async_channel::unbounded(); GlobalWatcher { - state: Mutex::new(WatcherState { + state: Arc::new(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( native_watcher.map(|watcher| { Box::new(SharedFakeWatchBackend(watcher)) as Box @@ -1221,6 +1360,7 @@ mod tests { }), ), event_tx, + _dispatch_task: Task::ready(()), } } @@ -1263,7 +1403,13 @@ mod tests { let (tx, rx) = async_channel::unbounded(); let pending_path_events: Arc>> = Default::default(); - let watcher = FsWatcher::new(cx.executor(), tx, pending_path_events.clone()); + let watcher = FsWatcher::new( + GlobalWatcher::new(cx.executor()), + crate::RealFs::new(None, cx.executor()), + cx.executor(), + tx, + pending_path_events.clone(), + ); watcher .add(&path) @@ -1582,6 +1728,42 @@ mod tests { ); } + #[test] + fn queued_rescans_naming_distinct_paths_are_all_dispatched() { + let (watcher, fired) = recording_watcher(); + let (event_tx, event_rx) = async_channel::unbounded(); + let rescan_of = |path: &str| { + notify::Event::new(EventKind::Other) + .set_flag(notify::event::Flag::Rescan) + .add_path(PathBuf::from(path)) + }; + + event_tx + .try_send((WatcherMode::Native, Ok(rescan_of("/repo/b")))) + .unwrap(); + watcher.dispatch_batch( + (WatcherMode::Native, Ok(rescan_of("/repo/a/nested"))), + &event_rx, + ); + + // Each named rescan only covers its own subtree once it reaches the + // callbacks, so collapsing the second one would lose /repo/b's resync. + // Both are broadcast to every registration of the mode. + 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/a/nested".to_owned(), + "/repo/b".to_owned(), + "/repo/b".to_owned(), + ] + ); + } + #[test] #[cfg(target_os = "windows")] fn event_dispatches_when_reported_path_has_verbatim_prefix() { @@ -1675,9 +1857,3 @@ mod tests { } } } - -pub fn global(f: impl FnOnce(&GlobalWatcher) -> T) -> anyhow::Result { - let global_watcher = global_watcher(); - global_watcher.ensure_native_watcher()?; - Ok(f(global_watcher)) -} diff --git a/crates/project/src/agent_server_store.rs b/crates/project/src/agent_server_store.rs index 63806e9b7c..6756319e10 100644 --- a/crates/project/src/agent_server_store.rs +++ b/crates/project/src/agent_server_store.rs @@ -1713,7 +1713,7 @@ mod tests { http_client: Arc, sha256: Option, ) -> LocalRegistryArchiveAgent { - let fs: Arc = Arc::new(fs::RealFs::new(None, cx.executor())); + let fs: Arc = fs::RealFs::new(None, cx.executor()); let target = RegistryTargetConfig { archive: TEST_ARCHIVE_URL.to_string(), cmd: "./agent".to_string(), diff --git a/crates/project/src/project.rs b/crates/project/src/project.rs index 3e8c279366..7f46ab17f0 100644 --- a/crates/project/src/project.rs +++ b/crates/project/src/project.rs @@ -2058,7 +2058,7 @@ impl Project { ) -> Entity { use clock::FakeSystemClock; - let fs = Arc::new(RealFs::new(None, cx.background_executor().clone())); + let fs = RealFs::new(None, cx.background_executor().clone()); let languages = LanguageRegistry::test(cx.background_executor().clone()); let clock = Arc::new(FakeSystemClock::new()); let http_client = http_client::FakeHttpClient::with_404_response(); diff --git a/crates/project/tests/integration/project_tests.rs b/crates/project/tests/integration/project_tests.rs index 5acd7d2536..7fd83b2e62 100644 --- a/crates/project/tests/integration/project_tests.rs +++ b/crates/project/tests/integration/project_tests.rs @@ -221,7 +221,7 @@ async fn test_symlinks(cx: &mut gpui::TestAppContext) { .unwrap(); let project = Project::test( - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), [root_link_path.as_ref()], cx, ) @@ -8607,7 +8607,7 @@ async fn test_rescan_and_remote_updates(cx: &mut gpui::TestAppContext) { } })); - let project = Project::test(Arc::new(RealFs::new(None, cx.executor())), [dir.path()], cx).await; + let project = Project::test(RealFs::new(None, cx.executor()), [dir.path()], cx).await; let buffer_for_path = |path: &'static str, cx: &mut gpui::TestAppContext| { let buffer = project.update(cx, |p, cx| p.open_local_buffer(dir.path().join(path), cx)); @@ -8761,7 +8761,7 @@ async fn test_recreated_directory_receives_child_events(cx: &mut gpui::TestAppCo cx.executor().allow_parking(); let dir = TempTree::new(json!({})); - let project = Project::test(Arc::new(RealFs::new(None, cx.executor())), [dir.path()], cx).await; + let project = Project::test(RealFs::new(None, cx.executor()), [dir.path()], cx).await; let tree = project.update(cx, |project, cx| project.worktrees(cx).next().unwrap()); tree.flush_fs_events(cx).await; @@ -16735,12 +16735,7 @@ async fn test_staging_hunk_preserve_executable_permission(cx: &mut gpui::TestApp git_commit("Initial commit", &repo); std::fs::write(&file_path, file_contents).unwrap(); - let project = Project::test( - Arc::new(RealFs::new(None, cx.executor())), - [root.path()], - cx, - ) - .await; + let project = Project::test(RealFs::new(None, cx.executor()), [root.path()], cx).await; let buffer = project .update(cx, |project, cx| { @@ -16975,12 +16970,7 @@ async fn test_git_repository_status(cx: &mut gpui::TestAppContext) { std::fs::remove_file(work_dir.join("d.txt")).unwrap(); std::fs::write(work_dir.join("a.txt"), "aa").unwrap(); - let project = Project::test( - Arc::new(RealFs::new(None, cx.executor())), - [root.path()], - cx, - ) - .await; + let project = Project::test(RealFs::new(None, cx.executor()), [root.path()], cx).await; let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap()); tree.flush_fs_events(cx).await; @@ -17262,12 +17252,7 @@ async fn test_git_events_after_project_excludes_dot_git(cx: &mut gpui::TestAppCo git_commit("Initial commit", &repo); git_branch("other-branch", &repo); - let project = Project::test( - Arc::new(RealFs::new(None, cx.executor())), - [work_dir.as_path()], - cx, - ) - .await; + let project = Project::test(RealFs::new(None, cx.executor()), [work_dir.as_path()], cx).await; let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap()); tree.flush_fs_events(cx).await; @@ -17354,12 +17339,7 @@ async fn test_git_status_postprocessing(cx: &mut gpui::TestAppContext) { // `sub` is a nested git repository. let _sub = git_init(&work_dir.join("sub")); - let project = Project::test( - Arc::new(RealFs::new(None, cx.executor())), - [root.path()], - cx, - ) - .await; + let project = Project::test(RealFs::new(None, cx.executor()), [root.path()], cx).await; let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap()); tree.flush_fs_events(cx).await; @@ -18210,7 +18190,7 @@ async fn test_conflicted_cherry_pick(cx: &mut gpui::TestAppContext) { git_add("a.txt", &repo); git_commit("init", &repo); - let project = Project::test(Arc::new(RealFs::new(None, cx.executor())), [root_path], cx).await; + let project = Project::test(RealFs::new(None, cx.executor()), [root_path], cx).await; let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap()); tree.flush_fs_events(cx).await; @@ -18366,7 +18346,7 @@ async fn test_rename_work_directory(cx: &mut gpui::TestAppContext) { git_commit("init", &repo); std::fs::write(root_path.join("projects/project1/a"), "aa").unwrap(); - let project = Project::test(Arc::new(RealFs::new(None, cx.executor())), [root_path], cx).await; + let project = Project::test(RealFs::new(None, cx.executor()), [root_path], cx).await; let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap()); tree.flush_fs_events(cx).await; @@ -18481,7 +18461,7 @@ async fn test_file_status(cx: &mut gpui::TestAppContext) { git_add(DOTGITIGNORE, &repo); git_commit("Initial commit", &repo); - let project = Project::test(Arc::new(RealFs::new(None, cx.executor())), [root_path], cx).await; + let project = Project::test(RealFs::new(None, cx.executor()), [root_path], cx).await; let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap()); tree.flush_fs_events(cx).await; @@ -18764,7 +18744,7 @@ async fn test_ignored_dirs_events(cx: &mut gpui::TestAppContext) { git_add(".gitignore", &repo); git_commit("Initial commit", &repo); - let project = Project::test(Arc::new(RealFs::new(None, cx.executor())), [root_path], cx).await; + let project = Project::test(RealFs::new(None, cx.executor()), [root_path], cx).await; let repository_updates = Arc::new(Mutex::new(Vec::new())); let project_events = Arc::new(Mutex::new(Vec::new())); project.update(cx, |project, cx| { @@ -20656,12 +20636,7 @@ async fn test_os_read_only_files_open_as_read_only(cx: &mut gpui::TestAppContext permissions.set_readonly(true); std::fs::set_permissions(&file_path, permissions).unwrap(); - let project = Project::test( - Arc::new(RealFs::new(None, cx.executor())), - [root.path()], - cx, - ) - .await; + let project = Project::test(RealFs::new(None, cx.executor()), [root.path()], cx).await; let buffer = project .update(cx, |project, cx| { diff --git a/crates/project_benchmarks/src/main.rs b/crates/project_benchmarks/src/main.rs index b2c4275487..60afe80659 100644 --- a/crates/project_benchmarks/src/main.rs +++ b/crates/project_benchmarks/src/main.rs @@ -136,7 +136,7 @@ fn main() -> Result<(), anyhow::Error> { let node = NodeRuntime::new(http_client, None, rx); let user_store = cx.new(|cx| UserStore::new(client.clone(), cx)); let registry = Arc::new(LanguageRegistry::new(cx.background_executor().clone())); - let fs = Arc::new(RealFs::new(None, cx.background_executor().clone())); + let fs = RealFs::new(None, cx.background_executor().clone()); diff --git a/crates/remote_server/src/server.rs b/crates/remote_server/src/server.rs index bd5bda5c26..6a4bf3d4d1 100644 --- a/crates/remote_server/src/server.rs +++ b/crates/remote_server/src/server.rs @@ -677,7 +677,7 @@ pub fn execute_run( json_schema_store::init(cx); let project = cx.new(|cx| { - let fs = Arc::new(RealFs::new(None, cx.background_executor().clone())); + let fs = RealFs::new(None, cx.background_executor().clone()); let node_settings_rx = initialize_settings(session.clone(), fs.clone(), cx); let proxy_url = read_proxy_settings(cx); diff --git a/crates/repl/src/repl_editor.rs b/crates/repl/src/repl_editor.rs index 61bed513a1..c9512b220c 100644 --- a/crates/repl/src/repl_editor.rs +++ b/crates/repl/src/repl_editor.rs @@ -839,7 +839,7 @@ mod tests { editor::init(cx); // Initialize the ReplStore with a fake filesystem - let fs = Arc::new(project::RealFs::new(None, cx.background_executor().clone())); + let fs = project::RealFs::new(None, cx.background_executor().clone()); ReplStore::init(fs, cx); // Add mock kernel specifications for TypeScript and Python diff --git a/crates/vim/src/test.rs b/crates/vim/src/test.rs index 6d5bbdbc11..7edaf44509 100644 --- a/crates/vim/src/test.rs +++ b/crates/vim/src/test.rs @@ -1432,7 +1432,7 @@ async fn test_remap(cx: &mut gpui::TestAppContext) { cx.update(|_, cx| { cx.bind_keys([KeyBinding::new( "g w", - workspace::SendKeystrokes(": j enter".to_string()), + workspace::SendKeystrokes(": j o i n space l i n e s enter".to_string()), None, )]) }); diff --git a/crates/workspace/src/multi_workspace.rs b/crates/workspace/src/multi_workspace.rs index 529e5b16c7..acd83f753c 100644 --- a/crates/workspace/src/multi_workspace.rs +++ b/crates/workspace/src/multi_workspace.rs @@ -1449,7 +1449,7 @@ impl MultiWorkspace { })); } - fn serialize_now(&mut self, cx: &mut Context) -> Task<()> { + fn serialize_now(&mut self, cx: &mut Context) -> impl Future + use<> { let state = MultiWorkspaceState { active_workspace_id: self.workspace().read(cx).database_id(), project_groups: self @@ -1467,16 +1467,17 @@ impl MultiWorkspace { }; let window_id = self.window_id; let kvp = db::kvp::KeyValueStore::global(cx); - cx.background_spawn(async move { + async move { crate::persistence::write_multi_workspace_state(&kvp, window_id, state).await; - }) + } } /// Used by the quit handler to ensure pending DB writes /// complete before the process exits. pub fn flush_serialization(&mut self, cx: &mut Context) -> Task<()> { self._serialize_task.take(); - self.serialize_now(cx) + let serialization = self.serialize_now(cx); + cx.spawn(async move |_, _| serialization.await) } pub fn flush_pending_serialization( diff --git a/crates/worktree/src/worktree.rs b/crates/worktree/src/worktree.rs index 9d218c7741..8037416e1e 100644 --- a/crates/worktree/src/worktree.rs +++ b/crates/worktree/src/worktree.rs @@ -1362,7 +1362,7 @@ impl LocalWorktree { let background = cx.background_executor().clone(); async move { let defer_watch = - force_defer_watch || (scanning_enabled && fs::requires_poll_watcher(&abs_path)); + force_defer_watch || (scanning_enabled && fs.requires_poll_watcher(&abs_path)); let (events, watcher) = if scanning_enabled && !defer_watch { fs.watch(&abs_path, FS_WATCH_LATENCY).await diff --git a/crates/worktree/tests/integration/worktree_tests.rs b/crates/worktree/tests/integration/worktree_tests.rs index 8f6476bbde..ffb4da9d3e 100644 --- a/crates/worktree/tests/integration/worktree_tests.rs +++ b/crates/worktree/tests/integration/worktree_tests.rs @@ -1208,7 +1208,7 @@ async fn test_real_fs_scan_symlinks_always(cx: &mut TestAppContext) { let tree = Worktree::local( project_root.as_path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -1270,7 +1270,7 @@ async fn test_real_fs_scan_symlinks_expanded(cx: &mut TestAppContext) { let tree = Worktree::local( project_root.as_path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -1435,7 +1435,7 @@ async fn test_renaming_case_only(cx: &mut TestAppContext) { const OLD_NAME: &str = "aaa.rs"; const NEW_NAME: &str = "AAA.rs"; - let fs = Arc::new(RealFs::new(None, cx.executor())); + let fs = RealFs::new(None, cx.executor()); let temp_root = TempTree::new(json!({ OLD_NAME: "", })); @@ -1553,9 +1553,7 @@ 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, -) { +async fn test_root_rescan_keeps_root_watcher_registered(cx: &mut TestAppContext) { init_test(cx); let fs = FakeFs::new(cx.background_executor.clone()); fs.insert_tree("/root", json!({})).await; @@ -1575,16 +1573,17 @@ async fn test_root_rescan_does_not_miss_event_before_readding_root_watcher( 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"); + // Dropping and re-registering the root watch would open a window in + // which filesystem events are lost. fs.emit_fs_event("/root", Some(PathEventKind::Rescan)); + tree.flush_fs_events(cx).await; - wait_for_condition(cx, |cx| { - tree.read_with(cx, |tree, _| { - tree.entry_for_path(rel_path("created-before-root-readd.txt")) - .is_some() - }) - }) - .await; + let root_watch_calls = fs + .watch_calls() + .into_iter() + .filter(|path| path == Path::new("/root")) + .count(); + assert_eq!(root_watch_calls, 1); } #[gpui::test] @@ -1983,7 +1982,7 @@ async fn test_write_file(cx: &mut TestAppContext) { let worktree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -1992,9 +1991,6 @@ async fn test_write_file(cx: &mut TestAppContext) { .await .unwrap(); - #[cfg(not(target_os = "macos"))] - fs::fs_watcher::global(|_| {}).unwrap(); - cx.read(|cx| worktree.read(cx).as_local().unwrap().scan_complete()) .await; worktree.flush_fs_events(cx).await; @@ -2082,7 +2078,7 @@ async fn test_file_scan_inclusions(cx: &mut TestAppContext) { let tree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -2151,7 +2147,7 @@ async fn test_file_scan_exclusions_overrules_inclusions(cx: &mut TestAppContext) let tree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -2213,7 +2209,7 @@ async fn test_file_scan_inclusions_reindexes_on_setting_change(cx: &mut TestAppC let tree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -2302,7 +2298,7 @@ async fn test_file_scan_exclusions(cx: &mut TestAppContext) { let tree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -2389,7 +2385,7 @@ async fn test_hidden_files(cx: &mut TestAppContext) { let tree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -2501,7 +2497,7 @@ async fn test_fs_events_in_exclusions(cx: &mut TestAppContext) { let tree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -2618,7 +2614,7 @@ async fn test_fs_events_in_dot_git_worktree(cx: &mut TestAppContext) { let tree = Worktree::local( dot_git_worktree_dir.clone(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), @@ -2775,7 +2771,7 @@ async fn test_create_dir_all_on_create_entry(cx: &mut TestAppContext) { assert!(tree.entry_for_path(rel_path("a/b")).unwrap().is_dir()); }); - let fs_real = Arc::new(RealFs::new(None, cx.executor())); + let fs_real = RealFs::new(None, cx.executor()); let temp_root = TempTree::new(json!({ "a": {} })); @@ -5425,7 +5421,7 @@ async fn test_ref_updates_in_dot_git_subdirectories_are_detected(cx: &mut TestAp let tree = Worktree::local( dir.path(), true, - Arc::new(RealFs::new(None, cx.executor())), + RealFs::new(None, cx.executor()), Default::default(), true, WorktreeId::from_proto(0), diff --git a/crates/worktree_benchmarks/src/main.rs b/crates/worktree_benchmarks/src/main.rs index 079867b5e8..87fa16f1c4 100644 --- a/crates/worktree_benchmarks/src/main.rs +++ b/crates/worktree_benchmarks/src/main.rs @@ -18,7 +18,7 @@ fn main() { app.run(|cx| { settings::init(cx); - let fs = Arc::new(RealFs::new(None, cx.background_executor().clone())); + let fs = RealFs::new(None, cx.background_executor().clone()); cx.spawn(async move |cx| { let worktree = Worktree::local( diff --git a/crates/zed/src/main.rs b/crates/zed/src/main.rs index c7e4eb6e91..4c2395dcd2 100644 --- a/crates/zed/src/main.rs +++ b/crates/zed/src/main.rs @@ -429,7 +429,7 @@ fn main() { log::info!("Using git binary path: {:?}", git_binary_path); } - let fs = Arc::new(RealFs::new(git_binary_path, app.background_executor())); + let fs = RealFs::new(git_binary_path, app.background_executor()); let (user_keymap_file_rx, user_keymap_watcher) = watch_config_file( &app.background_executor(), fs.clone(), diff --git a/crates/zed/src/visual_test_runner.rs b/crates/zed/src/visual_test_runner.rs index cea858e72e..6541bf666f 100644 --- a/crates/zed/src/visual_test_runner.rs +++ b/crates/zed/src/visual_test_runner.rs @@ -975,7 +975,7 @@ fn init_app_state(cx: &mut App) -> Arc { } // Use the real filesystem instead of FakeFs so we can access actual files on disk - let fs: Arc = Arc::new(fs::RealFs::new(None, cx.background_executor().clone())); + let fs: Arc = fs::RealFs::new(None, cx.background_executor().clone()); ::set_global(fs.clone(), cx); let languages = Arc::new(language::LanguageRegistry::test( diff --git a/crates/zed/src/zed.rs b/crates/zed/src/zed.rs index 259ea57f01..2bce5d08b4 100644 --- a/crates/zed/src/zed.rs +++ b/crates/zed/src/zed.rs @@ -572,7 +572,7 @@ pub fn initialize_workspace(app_state: Arc, cx: &mut App) { .detach(); #[cfg(not(any(test, target_os = "macos")))] - initialize_file_watcher(window, cx); + initialize_file_watcher(workspace.app_state().fs.as_ref(), window, cx); if let Some(specs) = window.gpu_specs() { log::info!("Using GPU: {:?}", specs); @@ -663,8 +663,8 @@ pub fn initialize_workspace(app_state: Arc, cx: &mut App) { #[cfg(any(target_os = "linux", target_os = "freebsd"))] #[allow(unused)] -fn initialize_file_watcher(window: &mut Window, cx: &mut Context) { - if let Err(e) = fs::fs_watcher::global(|_| {}) { +fn initialize_file_watcher(fs: &dyn Fs, window: &mut Window, cx: &mut Context) { + if let Err(e) = fs.start_native_watcher() { let message = format!( db::indoc! {r#" inotify_init returned {} @@ -694,8 +694,8 @@ fn initialize_file_watcher(window: &mut Window, cx: &mut Context) { #[cfg(target_os = "windows")] #[allow(unused)] -fn initialize_file_watcher(window: &mut Window, cx: &mut Context) { - if let Err(e) = fs::fs_watcher::global(|_| {}) { +fn initialize_file_watcher(fs: &dyn Fs, window: &mut Window, cx: &mut Context) { + if let Err(e) = fs.start_native_watcher() { let message = format!( db::indoc! {r#" ReadDirectoryChangesW initialization failed: {} @@ -8258,6 +8258,7 @@ mod tests { ) .await .expect("failed to save disable_ai=true to the fake settings file"); + executor.advance_clock(Duration::from_secs(2)); executor.run_until_parked(); cx.update(|cx| {