yenru0/zed
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fs: Refactor the code around GlobalWatcher to be more testable (#63894)

Browse Source 
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
This commit is contained in:
Cole Miller authored and GitHub committed 2026-09-07 19:01:41 +00:00
1 parent 43b8cd612f
commit 102a6c5d8a
23 files changed
+582 -448
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/agent/src/agent.rs
+1 -1
View File
@@ -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));
crates/component_preview/examples/component_preview.rs
+1 -1
View File
@@ -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());
<dyn fs::Fs>::set_global(fs.clone(), cx);
settings::init(cx);
crates/edit_prediction_cli/src/headless.rs
+1 -4
View File
@@ -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());
crates/eval_cli/src/headless.rs
+1 -4
View File
@@ -67,10 +67,7 @@ pub fn init(cx: &mut App) -> Arc<AgentCliAppState> {
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());
<dyn fs::Fs>::set_global(fs.clone(), cx);
let mut languages = LanguageRegistry::new(cx.background_executor().clone());
crates/extension_cli/src/main.rs
+1 -1
View File
@@ -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)?;
crates/extension_host/benches/extension_compilation_benchmark.rs
+2 -6
View File
@@ -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<WasmHost> {
});
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(
crates/extension_host/src/extension_store_test.rs
+1 -1
View File
@@ -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": {}
crates/fs/src/fs.rs
+194 -199
View File
@@ -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<Pin<Box<dyn Send + Stream<Item = Result<PathBuf>>>>>;
/// 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<Self>,
bundled_git_binary_path: Option<PathBuf>,
executor: BackgroundExecutor,
global_watcher: Arc<fs_watcher::GlobalWatcher>,
next_job_id: Arc<AtomicUsize>,
job_event_subscribers: Arc<Mutex<Vec<JobEventSender>>>,
trash: Arc<Mutex<SlotMap<TrashId, TrashedEntry>>>,
@@ -537,15 +552,17 @@ impl FileHandle for std::fs::File {
pub struct RealWatcher {}
impl RealFs {
pub fn new(git_binary_path: Option<PathBuf>, executor: BackgroundExecutor) -> Self {
Self {
pub fn new(git_binary_path: Option<PathBuf>, executor: BackgroundExecutor) -> Arc<Self> {
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<Box<dyn Send + Stream<Item = Vec<PathEvent>>>>,
Arc<dyn Watcher>,
) {
use util::{ResultExt as _, paths::SanitizedPath};
let executor = self.executor.clone();
let (tx, rx) = async_channel::unbounded();
let pending_paths: Arc<Mutex<Vec<PathEvent>>> = Default::default();
let watcher: Arc<dyn Watcher> = 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<Mutex<FakeFsState>>,
executor: gpui::BackgroundExecutor,
global_watcher: Arc<fs_watcher::GlobalWatcher>,
}
#[cfg(feature = "test-support")]
@@ -1438,7 +1428,8 @@ struct FakeFsState {
next_inode: u64,
next_mtime: SystemTime,
git_event_tx: async_channel::Sender<PathBuf>,
event_txs: Vec<(PathBuf, async_channel::Sender<Vec<PathEvent>>)>,
watch_roots: Vec<(PathBuf, std::sync::Weak<dyn Watcher>)>,
watches: FakeWatches,
events_paused: bool,
buffered_events: Vec<PathEvent>,
metadata_call_count: usize,
@@ -1446,41 +1437,19 @@ struct FakeFsState {
path_write_counts: std::collections::HashMap<PathBuf, usize>,
job_event_subscribers: Arc<Mutex<Vec<JobEventSender>>>,
trash: Mutex<SlotMap<TrashId, (TrashedEntry, FakeFsEntry)>>,
file_to_create_before_watch_add: Option<(PathBuf, PathBuf)>,
remove_dir_errors: std::collections::HashMap<PathBuf, String>,
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<PathBuf>,
watch_calls: Vec<PathBuf>,
event_sink: Option<Box<dyn Fn(notify::Result<notify::Event>) + 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::<Vec<_>>();
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(&registered_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<Mutex<FakeFsState>>,
}
#[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<Self> {
let (tx, rx) = async_channel::bounded::<PathBuf>(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>,
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()),
));
/// Every path the watcher backend has been asked to watch, in order,
/// including paths that were later unwatched.
pub fn watch_calls(&self) -> Vec<PathBuf> {
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<PathBuf> {
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<Vec<PathEvent>>,
fs_state: Arc<Mutex<FakeFsState>>,
prefixes: Mutex<Vec<PathBuf>>,
}
#[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<dyn Watcher>,
) {
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
}
crates/fs/src/fs_watcher.rs
+324 -148
View File
@@ -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<dyn Fs>,
global_watcher: Arc<GlobalWatcher>,
executor: BackgroundExecutor,
path: &Path,
latency: Duration,
) -> (
Pin<Box<dyn Send + Stream<Item = Vec<PathEvent>>>>,
Arc<dyn Watcher>,
) {
let (tx, rx) = async_channel::unbounded();
let pending_paths: Arc<Mutex<Vec<PathEvent>>> = Default::default();
let watcher: Arc<dyn Watcher> = 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<GlobalWatcher>,
fs: Arc<dyn Fs>,
executor: BackgroundExecutor,
tx: async_channel::Sender<()>,
pending_path_events: Arc<Mutex<Vec<PathEvent>>>,
@@ -35,12 +106,16 @@ struct FsWatcherRegistration {
}
impl FsWatcher {
pub fn new(
pub(crate) fn new(
global_watcher: Arc<GlobalWatcher>,
fs: Arc<dyn Fs>,
executor: BackgroundExecutor,
tx: async_channel::Sender<()>,
pending_path_events: Arc<Mutex<Vec<PathEvent>>>,
) -> Self {
Self {
global_watcher,
fs,
executor,
tx,
pending_path_events,
@@ -50,13 +125,15 @@ impl FsWatcher {
}
fn add_existing_path(&self, path: Arc<Path>) -> 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<Path>,
case_insensitive: bool,
tx: async_channel::Sender<()>,
pending_path_events: Arc<Mutex<Vec<PathEvent>>>,
) -> anyhow::Result<Option<FsWatcherRegistration>> {
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<GlobalWatcher>,
fs: Arc<dyn Fs>,
executor: BackgroundExecutor,
path: Arc<Path>,
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<T: notify::Watcher + Send> WatchBackend for T {
type DispatchEvent = (WatcherMode, Result<notify::Event, notify::Error>);
pub struct GlobalWatcher {
state: Mutex<WatcherState>,
state: Arc<Mutex<WatcherState>>,
// DANGER: never keep state lock while holding watcher lock
// two mutexes because calling watcher.add triggers watcher.event, which needs watchers.
native_watcher: Mutex<Option<Box<dyn WatchBackend>>>,
poll_watcher: Mutex<Option<Box<dyn WatchBackend>>>,
event_tx: async_channel::Sender<DispatchEvent>,
_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> {
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<Box<dyn WatchBackend>>,
) -> Arc<Self> {
let (event_tx, event_rx) = async_channel::unbounded::<DispatchEvent>();
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<notify::Event>) + 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<notify::Event, notify::Error>) {
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<notify::Event, notify::Error>) {
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::<Vec<_>>();
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::<Vec<_>>()
}
};
for callback in callbacks {
callback(&event);
}
dispatch(&self.state, mode, event);
}
#[cfg(test)]
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);
}
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 = <notify::RecommendedWatcher as notify::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<WatcherState>,
mode: WatcherMode,
event: Result<notify::Event, notify::Error>,
) {
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::<Vec<_>>();
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::<Vec<_>>()
}
};
for callback in callbacks {
callback(&event);
}
}
fn dispatch_batch(
state: &Mutex<WatcherState>,
first: DispatchEvent,
event_rx: &async_channel::Receiver<DispatchEvent>,
) {
// 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<DispatchEvent>,
mode: WatcherMode,
event: Result<notify::Event, notify::Error>,
) {
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::<notify::Error>()
@@ -1109,34 +1276,6 @@ pub fn poll_interval() -> Duration {
*POLL_INTERVAL
}
static FS_WATCHER_INSTANCE: OnceLock<GlobalWatcher> = OnceLock::new();
fn global_watcher() -> &'static GlobalWatcher {
FS_WATCHER_INSTANCE.get_or_init(|| {
let (event_tx, event_rx) = async_channel::unbounded::<DispatchEvent>();
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<dyn WatchBackend>
@@ -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<Mutex<Vec<PathEvent>>> = 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<T>(f: impl FnOnce(&GlobalWatcher) -> T) -> anyhow::Result<T> {
let global_watcher = global_watcher();
global_watcher.ensure_native_watcher()?;
Ok(f(global_watcher))
}
crates/project/src/agent_server_store.rs
+1 -1
View File
@@ -1713,7 +1713,7 @@ mod tests {
http_client: Arc<dyn HttpClient>,
sha256: Option<String>,
) -> LocalRegistryArchiveAgent {
let fs: Arc<dyn Fs> = Arc::new(fs::RealFs::new(None, cx.executor()));
let fs: Arc<dyn Fs> = fs::RealFs::new(None, cx.executor());
let target = RegistryTargetConfig {
archive: TEST_ARCHIVE_URL.to_string(),
cmd: "./agent".to_string(),
crates/project/src/project.rs
+1 -1
View File
@@ -2058,7 +2058,7 @@ impl Project {
) -> Entity<Project> {
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();
crates/project/tests/integration/project_tests.rs
+12 -37
View File
@@ -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| {
crates/project_benchmarks/src/main.rs
+1 -1
View File
@@ -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());
crates/remote_server/src/server.rs
+1 -1
View File
@@ -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);
crates/repl/src/repl_editor.rs
+1 -1
View File
@@ -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
crates/vim/src/test.rs
+1 -1
View File
@@ -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,
)])
});
crates/workspace/src/multi_workspace.rs
+5 -4
View File
@@ -1449,7 +1449,7 @@ impl MultiWorkspace {
}));
}
fn serialize_now(&mut self, cx: &mut Context<Self>) -> Task<()> {
fn serialize_now(&mut self, cx: &mut Context<Self>) -> impl Future<Output = ()> + 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<Self>) -> 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(
crates/worktree/src/worktree.rs
+1 -1
View File
@@ -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
crates/worktree/tests/integration/worktree_tests.rs
+23 -27
View File
@@ -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),
crates/worktree_benchmarks/src/main.rs
+1 -1
View File
@@ -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(
crates/zed/src/main.rs
+1 -1
View File
@@ -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(),
crates/zed/src/visual_test_runner.rs
+1 -1
View File
@@ -975,7 +975,7 @@ fn init_app_state(cx: &mut App) -> Arc<AppState> {
}
// Use the real filesystem instead of FakeFs so we can access actual files on disk
let fs: Arc<dyn Fs> = Arc::new(fs::RealFs::new(None, cx.background_executor().clone()));
let fs: Arc<dyn Fs> = fs::RealFs::new(None, cx.background_executor().clone());
<dyn Fs>::set_global(fs.clone(), cx);
let languages = Arc::new(language::LanguageRegistry::test(
crates/zed/src/zed.rs
+6 -5
View File
@@ -572,7 +572,7 @@ pub fn initialize_workspace(app_state: Arc<AppState>, 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<AppState>, cx: &mut App) {
#[cfg(any(target_os = "linux", target_os = "freebsd"))]
#[allow(unused)]
fn initialize_file_watcher(window: &mut Window, cx: &mut Context<Workspace>) {
if let Err(e) = fs::fs_watcher::global(|_| {}) {
fn initialize_file_watcher(fs: &dyn Fs, window: &mut Window, cx: &mut Context<Workspace>) {
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<Workspace>) {
#[cfg(target_os = "windows")]
#[allow(unused)]
fn initialize_file_watcher(window: &mut Window, cx: &mut Context<Workspace>) {
if let Err(e) = fs::fs_watcher::global(|_| {}) {
fn initialize_file_watcher(fs: &dyn Fs, window: &mut Window, cx: &mut Context<Workspace>) {
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| {