tenant.openagents/omega

1use gpui::{BackgroundExecutor, Task};
2use notify::{Event, EventKind};
3use parking_lot::Mutex;
4use std::{
5    collections::HashMap,
6    fs,
7    ops::DerefMut,
8    path::Path,
9    sync::{Arc, LazyLock, OnceLock},
10    time::{Duration, Instant},
11};
12use util::{ResultExt, paths::SanitizedPath};
13
14use crate::{PathEvent, PathEventKind, Watcher};
15
16#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
17pub enum WatcherMode {
18    #[default]
19    Native,
20    Poll,
21}
22
23pub struct FsWatcher {
24    executor: BackgroundExecutor,
25    tx: async_channel::Sender<()>,
26    pending_path_events: Arc<Mutex<Vec<PathEvent>>>,
27    registrations: Arc<Mutex<HashMap<WatchKey, FsWatcherRegistration>>>,
28    pending_registrations: Arc<Mutex<HashMap<Arc<std::path::Path>, Task<()>>>>,
29}
30
31#[derive(Clone, Copy)]
32struct FsWatcherRegistration {
33    id: WatcherRegistrationId,
34    mode: WatcherMode,
35}
36
37impl FsWatcher {
38    pub fn new(
39        executor: BackgroundExecutor,
40        tx: async_channel::Sender<()>,
41        pending_path_events: Arc<Mutex<Vec<PathEvent>>>,
42    ) -> Self {
43        Self {
44            executor,
45            tx,
46            pending_path_events,
47            registrations: Default::default(),
48            pending_registrations: Default::default(),
49        }
50    }
51
52    fn add_existing_path(&self, path: Arc<Path>) -> anyhow::Result<()> {
53        let case_insensitive = case_insensitive_path(&path);
54        let key = WatchKey::for_registration(SanitizedPath::new(&path), case_insensitive);
55        if self.registrations.lock().contains_key(&key) {
56            log::trace!("path to watch is already watched: {path:?}");
57            return Ok(());
58        }
59        match register_existing_path(
60            path.clone(),
61            case_insensitive,
62            self.tx.clone(),
63            self.pending_path_events.clone(),
64        )? {
65            Some(registration) => {
66                self.registrations.lock().insert(key, registration);
67            }
68            None => {
69                // Registration was skipped (e.g. the native watch-limit cooldown
70                // is active). Retry in the background rather than silently leaving
71                // the path unwatched forever.
72                log::warn!("watch registration for {path:?} was skipped; retrying in background");
73                self.add_pending_path(path);
74            }
75        }
76        Ok(())
77    }
78
79    fn add_pending_path(&self, path: Arc<Path>) {
80        let mut pending_registrations = self.pending_registrations.lock();
81        if pending_registrations.contains_key(path.as_ref()) {
82            return;
83        }
84
85        let task = self.executor.spawn(poll_path_until_created(
86            self.executor.clone(),
87            path.clone(),
88            self.tx.clone(),
89            self.pending_path_events.clone(),
90            self.registrations.clone(),
91            self.pending_registrations.clone(),
92        ));
93        pending_registrations.insert(path, task);
94    }
95}
96
97impl Drop for FsWatcher {
98    fn drop(&mut self) {
99        self.pending_registrations.lock().clear();
100
101        let mut registrations = HashMap::new();
102        {
103            let old = &mut self.registrations.lock();
104            std::mem::swap(old.deref_mut(), &mut registrations);
105        }
106
107        let global_watcher = global_watcher();
108        for (_, registration) in registrations {
109            global_watcher.remove(registration.id);
110        }
111    }
112}
113
114impl Watcher for FsWatcher {
115    fn add(&self, path: &std::path::Path) -> anyhow::Result<()> {
116        log::trace!("watcher add: {path:?}");
117
118        let path: Arc<Path> = path.into();
119        if path_covered_by_recursive_registration(
120            &self.registrations.lock(),
121            SanitizedPath::new(&path),
122        ) {
123            log::trace!("path to watch is covered by an existing registration: {path:?}");
124            return Ok(());
125        }
126
127        if self
128            .pending_registrations
129            .lock()
130            .contains_key(path.as_ref())
131        {
132            log::trace!("path to watch is already pending: {path:?}");
133            return Ok(());
134        }
135
136        if fs::symlink_metadata(path.as_ref()).is_err() {
137            self.add_pending_path(path);
138            return Ok(());
139        }
140
141        self.add_existing_path(path)
142    }
143
144    fn remove(&self, path: &std::path::Path) -> anyhow::Result<()> {
145        log::trace!("remove watched path: {path:?}");
146        self.pending_registrations.lock().remove(path);
147
148        let sanitized = SanitizedPath::new(path);
149        let registration = {
150            let mut registrations = self.registrations.lock();
151            registrations
152                .remove(&WatchKey::exact(sanitized))
153                .or_else(|| registrations.remove(&WatchKey::folded(sanitized)))
154        };
155        if let Some(registration) = registration {
156            global_watcher().remove(registration.id);
157        }
158        Ok(())
159    }
160}
161
162/// Whether a recursive registration on a strict ancestor of `path` already covers
163/// it. Both key spellings are probed so a folded registration still matches; only
164/// poll watches and native macOS/Windows watches are recursive.
165fn path_covered_by_recursive_registration(
166    registrations: &HashMap<WatchKey, FsWatcherRegistration>,
167    path: &SanitizedPath,
168) -> bool {
169    path.as_path().ancestors().skip(1).any(|ancestor| {
170        let ancestor = SanitizedPath::unchecked_new(ancestor);
171        [WatchKey::exact(ancestor), WatchKey::folded(ancestor)]
172            .iter()
173            .any(|key| {
174                registrations.get(key).is_some_and(|registration| {
175                    registration.mode == WatcherMode::Poll
176                        || cfg!(any(target_os = "windows", target_os = "macos"))
177                })
178            })
179    })
180}
181
182/// Detect whether a path requires polling instead of native file watching.
183///
184/// Returns `true` for filesystem types where inotify/FSEvents/ReadDirectoryChanges
185/// silently fail to deliver events: 9P (WSL drvfs), NFS, CIFS/SMB, FUSE (sshfs), etc.
186///
187/// Can be overridden with the `ZED_FILE_WATCHER_MODE` environment variable:
188/// - `native` — always use native OS watcher
189/// - `poll` — always use polling
190/// - `auto` (default) — auto-detect based on filesystem type
191pub fn requires_poll_watcher(path: &Path) -> bool {
192    match std::env::var("ZED_FILE_WATCHER_MODE")
193        .as_deref()
194        .unwrap_or("auto")
195    {
196        "native" => return false,
197        "poll" => return true,
198        _ => {}
199    }
200
201    #[cfg(target_os = "linux")]
202    {
203        return detect_requires_poll_watcher_linux(path);
204    }
205
206    #[cfg(not(target_os = "linux"))]
207    {
208        let _ = path;
209        false
210    }
211}
212
213fn register_existing_path(
214    path: Arc<Path>,
215    case_insensitive: bool,
216    tx: async_channel::Sender<()>,
217    pending_path_events: Arc<Mutex<Vec<PathEvent>>>,
218) -> anyhow::Result<Option<FsWatcherRegistration>> {
219    let mode = if requires_poll_watcher(path.as_ref()) {
220        log::info!(
221            "Using poll watcher ({}ms interval) for {}",
222            poll_interval().as_millis(),
223            path.display()
224        );
225        telemetry::event!("fs_watcher_poll", path = path.display().to_string());
226        WatcherMode::Poll
227    } else {
228        WatcherMode::Native
229    };
230    let root_path = SanitizedPath::new_arc(path.as_ref());
231    let path_for_callback = path.clone();
232    let Some(registration_id) = global_watcher().add(
233        path,
234        mode,
235        case_insensitive,
236        move |event: &notify::Event| {
237            log::trace!("watcher received event: {event:?}");
238            push_notify_event(
239                &tx,
240                &pending_path_events,
241                &root_path,
242                case_insensitive,
243                path_for_callback.as_ref(),
244                event,
245            );
246        },
247    )?
248    else {
249        return Ok(None);
250    };
251    Ok(Some(FsWatcherRegistration {
252        id: registration_id,
253        mode,
254    }))
255}
256
257#[cfg(target_os = "linux")]
258fn detect_requires_poll_watcher_linux(path: &Path) -> bool {
259    use std::ffi::CString;
260    use std::os::unix::ffi::OsStrExt;
261
262    let c_path = match CString::new(path.as_os_str().as_bytes()) {
263        Ok(p) => p,
264        Err(_) => return false,
265    };
266
267    let mut stat: libc::statfs = unsafe { std::mem::zeroed() };
268    if unsafe { libc::statfs(c_path.as_ptr(), &mut stat) } != 0 {
269        return false;
270    }
271
272    const V9FS_MAGIC: u64 = 0x0102_1997;
273    const NFS_SUPER_MAGIC: u64 = 0x0000_6969;
274    const CIFS_MAGIC: u64 = 0xFF53_4D42;
275    const SMB_SUPER_MAGIC: u64 = 0x0000_517B;
276    const SMB2_MAGIC: u64 = 0xFE53_4D42;
277    const FUSE_SUPER_MAGIC: u64 = 0x6573_5546;
278
279    let fs_type = (stat.f_type as u64) & 0xFFFF_FFFF;
280    if fs_type == FUSE_SUPER_MAGIC && is_virtiofs(path) {
281        return false;
282    }
283
284    if fs_type == V9FS_MAGIC
285        || fs_type == NFS_SUPER_MAGIC
286        || fs_type == CIFS_MAGIC
287        || fs_type == SMB_SUPER_MAGIC
288        || fs_type == SMB2_MAGIC
289        || fs_type == FUSE_SUPER_MAGIC
290    {
291        log::info!(
292            "Detected network/virtual filesystem (type 0x{:x}) at {}, using poll watcher",
293            fs_type,
294            path.display()
295        );
296        return true;
297    }
298
299    if is_wsl_drvfs_path(path) {
300        log::info!(
301            "Detected WSL drvfs mount at {}, using poll watcher",
302            path.display()
303        );
304        return true;
305    }
306
307    false
308}
309
310#[cfg(target_os = "linux")]
311fn is_virtiofs(path: &Path) -> bool {
312    let Ok(mountinfo) = std::fs::read_to_string("/proc/self/mountinfo") else {
313        return false;
314    };
315
316    let mut best_mount = None;
317    for line in mountinfo.lines() {
318        let fields = line.split(' ').collect::<Vec<_>>();
319        let Some(separator) = fields.iter().position(|field| *field == "-") else {
320            continue;
321        };
322        let (Some(mount_point), Some(fs_type)) = (fields.get(4), fields.get(separator + 1)) else {
323            continue;
324        };
325
326        let mount_point = mount_point
327            .replace("\\040", " ")
328            .replace("\\011", "\t")
329            .replace("\\012", "\n")
330            .replace("\\134", "\\");
331        if path.starts_with(&mount_point)
332            && best_mount.is_none_or(|(length, _)| mount_point.len() > length)
333        {
334            best_mount = Some((mount_point.len(), *fs_type));
335        }
336    }
337
338    best_mount.is_some_and(|(_, fs_type)| fs_type == "virtiofs" || fs_type == "fuse.virtiofs")
339}
340
341#[cfg(target_os = "linux")]
342fn is_wsl_drvfs_path(path: &Path) -> bool {
343    if std::env::var_os("WSL_DISTRO_NAME").is_none() {
344        if let Ok(version) = std::fs::read_to_string("/proc/version") {
345            let version = version.to_lowercase();
346            if !version.contains("microsoft") && !version.contains("wsl") {
347                return false;
348            }
349        } else {
350            return false;
351        }
352    }
353
354    let Some(path) = path.to_str() else {
355        return false;
356    };
357    if !path.starts_with("/mnt/") || path.len() < 6 {
358        return false;
359    }
360    let after_mnt = &path[5..];
361    after_mnt.starts_with(|c: char| c.is_ascii_alphabetic())
362        && (after_mnt.len() == 1 || after_mnt.as_bytes()[1] == b'/')
363}
364
365/// Whether the volume backing `path` does case-insensitive name lookups, used to
366/// pick exact vs. folded matching.
367#[cfg(target_os = "macos")]
368fn case_insensitive_path(path: &Path) -> bool {
369    use std::os::unix::ffi::OsStrExt as _;
370
371    // `pathconf(_PC_CASE_SENSITIVE)` returns 1 (sensitive), 0 (insensitive), or -1
372    // on error; default errors to insensitive (the APFS/HFS+ default).
373    let Ok(c_path) = std::ffi::CString::new(path.as_os_str().as_bytes()) else {
374        return true;
375    };
376    // SAFETY: We just initialized c_path, so it's a valid pointer
377    unsafe { libc::pathconf(c_path.as_ptr(), libc::_PC_CASE_SENSITIVE) == 0 }
378}
379
380#[cfg(target_os = "linux")]
381fn case_insensitive_path(_path: &Path) -> bool {
382    // use std::os::unix::ffi::OsStrExt as _;
383
384    // // Only ext4/f2fs casefold (`+F`) dirs are insensitive, reported by `statx` via
385    // // STATX_ATTR_CASEFOLD; any failure (e.g. pre-4.11 ENOSYS) means case-sensitive.
386    // const STATX_ATTR_CASEFOLD: u64 = 0x0000_2000;
387    // let Ok(c_path) = std::ffi::CString::new(path.as_os_str().as_bytes()) else {
388    //     return false;
389    // };
390    // let mut buf = std::mem::MaybeUninit::<libc::statx>::zeroed();
391
392    // // SAFETY: c_path is still valid, buffer has been zeroed
393    // if unsafe { libc::statx(libc::AT_FDCWD, c_path.as_ptr(), 0, 0, buf.as_mut_ptr()) } != 0 {
394    //     return false;
395    // }
396
397    // // SAFETY: libc statx initialized this buffer, otherwise we would've returned on a error
398    // // in that function call
399    // let buf = unsafe { buf.assume_init() };
400    // buf.stx_attributes_mask & STATX_ATTR_CASEFOLD != 0
401    //     && buf.stx_attributes & STATX_ATTR_CASEFOLD != 0
402    false
403}
404
405#[cfg(target_os = "windows")]
406fn case_insensitive_path(_path: &Path) -> bool {
407    // todo(windows): Windows defaults to case in sensitive, but
408    // they can mark specific directories as case sensitive. Mainly
409    // for WSL use cases
410    true
411}
412
413#[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))]
414fn case_insensitive_path(_path: &Path) -> bool {
415    // Other BSDs default to case-sensitive local filesystems.
416    false
417}
418
419/// Whether `path` is `root` or sits beneath it, folding case on case-insensitive
420/// volumes so a differently-cased spelling still matches.
421fn path_is_under(path: &SanitizedPath, root: &SanitizedPath, case_insensitive: bool) -> bool {
422    if case_insensitive {
423        let path = path.as_path().to_string_lossy().to_lowercase();
424        let root = root.as_path().to_string_lossy().to_lowercase();
425        Path::new(&path).starts_with(Path::new(&root))
426    } else {
427        path.starts_with(root)
428    }
429}
430
431/// Lookup key for a watch path, shared by add, remove, and dispatch so they all
432/// agree on whether two spellings denote the same directory.
433///
434/// On case-sensitive volumes the exact (sanitized) path is the key, so genuinely
435/// distinct directories stay distinct. On case-insensitive volumes the folded
436/// (lowercased) spelling is the key, so any casing of a directory collides.
437///
438/// The two variants are distinct map keys, so a case-sensitive registration can
439/// never be hit by a folded lookup (or vice versa); dispatch can therefore probe
440/// both forms of an event path without risking a cross-rule false match.
441///
442/// NOTE: folding only normalizes case. macOS (APFS/HFS+) is also Unicode
443/// normalization-insensitive (NFC vs NFD); that normalization is intentionally
444/// centralized here so it can be added in one place later without touching the
445/// add/remove/dispatch call sites.
446#[derive(Clone, Debug, PartialEq, Eq, Hash)]
447enum WatchKey {
448    Exact(Arc<Path>),
449    Folded(Arc<str>),
450}
451
452impl WatchKey {
453    fn exact(path: &SanitizedPath) -> Self {
454        Self::Exact(Arc::from(path.as_path()))
455    }
456
457    fn folded(path: &SanitizedPath) -> Self {
458        let lossy = path.as_path().to_string_lossy();
459        // macOS (APFS/HFS+) compares names normalization-insensitively (NFC vs
460        // NFD), and FSEvents can report NFD while a config/LSP supplies NFC, so
461        // normalize before folding case. Windows (NTFS) and Linux are
462        // normalization-sensitive, so there we only fold case.
463        #[cfg(target_os = "macos")]
464        let folded = {
465            use unicode_normalization::UnicodeNormalization as _;
466            lossy.chars().nfc().collect::<String>().to_lowercase()
467        };
468        #[cfg(not(target_os = "macos"))]
469        let folded = lossy.to_lowercase();
470        Self::Folded(folded.into())
471    }
472
473    fn for_registration(path: &SanitizedPath, case_insensitive: bool) -> Self {
474        if case_insensitive {
475            Self::folded(path)
476        } else {
477            Self::exact(path)
478        }
479    }
480}
481
482async fn poll_path_until_created(
483    executor: BackgroundExecutor,
484    path: Arc<Path>,
485    tx: async_channel::Sender<()>,
486    pending_path_events: Arc<Mutex<Vec<PathEvent>>>,
487    registrations: Arc<Mutex<HashMap<WatchKey, FsWatcherRegistration>>>,
488    pending_registrations: Arc<Mutex<HashMap<Arc<Path>, Task<()>>>>,
489) {
490    loop {
491        executor.timer(poll_interval()).await;
492
493        if !pending_registrations.lock().contains_key(path.as_ref()) {
494            return;
495        }
496
497        if smol::fs::symlink_metadata(path.as_ref()).await.is_err() {
498            continue;
499        }
500
501        // Probe case sensitivity now that the path exists, rather than at add
502        // time when it didn't.
503        let case_insensitive = case_insensitive_path(path.as_ref());
504        let key = WatchKey::for_registration(SanitizedPath::new(&path), case_insensitive);
505
506        if registrations.lock().contains_key(&key) {
507            pending_registrations.lock().remove(path.as_ref());
508            return;
509        }
510
511        match register_existing_path(
512            path.clone(),
513            case_insensitive,
514            tx.clone(),
515            pending_path_events.clone(),
516        ) {
517            Ok(Some(registration)) => {
518                {
519                    let mut pending_registrations = pending_registrations.lock();
520                    if pending_registrations.remove(path.as_ref()).is_none() {
521                        global_watcher().remove(registration.id);
522                        return;
523                    }
524                    registrations.lock().insert(key, registration);
525                }
526                enqueue_path_events(
527                    &tx,
528                    &pending_path_events,
529                    vec![
530                        PathEvent {
531                            path: path.to_path_buf(),
532                            kind: Some(PathEventKind::Created),
533                        },
534                        PathEvent {
535                            path: path.to_path_buf(),
536                            kind: Some(PathEventKind::Rescan),
537                        },
538                    ],
539                );
540                return;
541            }
542            Ok(None) => {}
543            Err(error) => {
544                log::warn!("failed to watch newly-created path {path:?}: {error}; retrying");
545            }
546        }
547    }
548}
549
550fn enqueue_path_events(
551    tx: &smol::channel::Sender<()>,
552    pending_path_events: &Arc<Mutex<Vec<PathEvent>>>,
553    mut path_events: Vec<PathEvent>,
554) {
555    if path_events.is_empty() {
556        return;
557    }
558
559    path_events.sort();
560    let mut pending_paths = pending_path_events.lock();
561    if pending_paths.is_empty() {
562        tx.try_send(()).ok();
563    }
564    coalesce_pending_rescans(&mut pending_paths, &mut path_events);
565    util::extend_sorted(&mut *pending_paths, path_events, usize::MAX, |a, b| {
566        a.path.cmp(&b.path)
567    });
568}
569
570fn push_notify_event(
571    tx: &smol::channel::Sender<()>,
572    pending_path_events: &Arc<Mutex<Vec<PathEvent>>>,
573    root_path: &SanitizedPath,
574    case_insensitive: bool,
575    watched_root: &Path,
576    event: &notify::Event,
577) {
578    let kind = match event.kind {
579        EventKind::Create(_) => Some(PathEventKind::Created),
580        EventKind::Modify(_) => Some(PathEventKind::Changed),
581        EventKind::Remove(_) => Some(PathEventKind::Removed),
582        _ => None,
583    };
584    let mut path_events = event
585        .paths
586        .iter()
587        .filter_map(|event_path| {
588            let event_path = SanitizedPath::new(event_path);
589            path_is_under(event_path, root_path, case_insensitive).then(|| PathEvent {
590                path: event_path.as_path().to_path_buf(),
591                kind,
592            })
593        })
594        .collect::<Vec<_>>();
595
596    if event.need_rescan() {
597        if !watcher_logging_rate_limited() {
598            log::warn!("filesystem watcher lost sync for {watched_root:?}; scheduling rescan");
599        }
600
601        path_events.retain(|path_event| path_event.path != watched_root);
602        path_events.push(PathEvent {
603            path: watched_root.to_path_buf(),
604            kind: Some(PathEventKind::Rescan),
605        });
606    }
607    log::trace!("path_events: {:?}", path_events);
608    enqueue_path_events(tx, pending_path_events, path_events);
609}
610
611fn watcher_logging_rate_limited() -> bool {
612    static LAST_WARN: Mutex<Option<(Instant, usize)>> = Mutex::new(None);
613    let Some((ref mut started, ref mut emitted)) = *LAST_WARN.lock() else {
614        *LAST_WARN.lock() = Some((Instant::now(), 0));
615        return false;
616    };
617
618    if started.elapsed().as_secs() < 1 {
619        if *emitted < 20 {
620            log::warn!("filesystem watcher lost sync for many files, not logging more");
621            return true;
622        } else {
623            *emitted += 1;
624        }
625    } else {
626        *emitted = 0;
627        *started = Instant::now()
628    }
629
630    true
631}
632
633fn coalesce_pending_rescans(pending_paths: &mut Vec<PathEvent>, path_events: &mut Vec<PathEvent>) {
634    if !path_events
635        .iter()
636        .any(|event| event.kind == Some(PathEventKind::Rescan))
637    {
638        return;
639    }
640
641    let mut new_rescan_paths: Vec<std::path::PathBuf> = path_events
642        .iter()
643        .filter(|e| e.kind == Some(PathEventKind::Rescan))
644        .map(|e| e.path.clone())
645        .collect();
646    new_rescan_paths.sort_unstable();
647
648    let mut deduped_rescans: Vec<std::path::PathBuf> = Vec::with_capacity(new_rescan_paths.len());
649    for path in new_rescan_paths {
650        if deduped_rescans
651            .iter()
652            .any(|ancestor| path != *ancestor && path.starts_with(ancestor))
653        {
654            continue;
655        }
656        deduped_rescans.push(path);
657    }
658
659    deduped_rescans.retain(|new_path| {
660        !pending_paths
661            .iter()
662            .any(|pending| is_covered_rescan(pending.kind, new_path, &pending.path))
663    });
664
665    if !deduped_rescans.is_empty() {
666        pending_paths.retain(|pending| {
667            !deduped_rescans.iter().any(|rescan_path| {
668                pending.path == *rescan_path
669                    || is_covered_rescan(pending.kind, &pending.path, rescan_path)
670            })
671        });
672    }
673
674    path_events.retain(|event| {
675        event.kind != Some(PathEventKind::Rescan) || deduped_rescans.contains(&event.path)
676    });
677}
678
679fn is_covered_rescan(kind: Option<PathEventKind>, path: &Path, ancestor: &Path) -> bool {
680    kind == Some(PathEventKind::Rescan) && path != ancestor && path.starts_with(ancestor)
681}
682
683#[derive(Default, Debug, Copy, Clone, PartialEq, Eq, Hash)]
684pub struct WatcherRegistrationId(u32);
685
686struct WatcherRegistrationState {
687    callback: Arc<dyn Fn(&notify::Event) + Send + Sync>,
688    key: WatchKey,
689    path: Arc<SanitizedPath>,
690    mode: WatcherMode,
691}
692
693struct PathRegistrationState {
694    watcher_ids: Vec<WatcherRegistrationId>,
695    has_os_watcher: bool,
696}
697
698/// The registered watch paths for one watcher mode, keyed by [`WatchKey`] so that
699/// add (dedup), remove, and dispatch share a single notion of path identity.
700#[derive(Default)]
701struct WatchPaths(HashMap<WatchKey, PathRegistrationState>);
702
703impl WatchPaths {
704    fn contains(&self, key: &WatchKey) -> bool {
705        self.0.contains_key(key)
706    }
707
708    fn get_mut(&mut self, key: &WatchKey) -> Option<&mut PathRegistrationState> {
709        self.0.get_mut(key)
710    }
711
712    fn entry(
713        &mut self,
714        key: WatchKey,
715    ) -> std::collections::hash_map::Entry<'_, WatchKey, PathRegistrationState> {
716        self.0.entry(key)
717    }
718
719    fn remove(&mut self, key: &WatchKey) {
720        self.0.remove(key);
721    }
722
723    /// True if a recursive registration on a strict ancestor already covers
724    /// `path`. Only poll watches and native macOS/Windows watches are recursive.
725    fn covered_by_recursive_ancestor(&self, path: &SanitizedPath, mode: WatcherMode) -> bool {
726        if mode != WatcherMode::Poll && !cfg!(any(target_os = "windows", target_os = "macos")) {
727            return false;
728        }
729        path.as_path().ancestors().skip(1).any(|ancestor| {
730            let ancestor = SanitizedPath::unchecked_new(ancestor);
731            self.0.contains_key(&WatchKey::exact(ancestor))
732                || self.0.contains_key(&WatchKey::folded(ancestor))
733        })
734    }
735
736    /// Collects the watcher ids of every registration whose directory is an
737    /// ancestor of (or equal to) `path`. Both exact and folded keys are probed,
738    /// so a real-cased event path matches a folded registration and vice versa.
739    fn watcher_ids_covering(&self, path: &SanitizedPath, ids: &mut Vec<WatcherRegistrationId>) {
740        for ancestor in path.as_path().ancestors() {
741            let ancestor = SanitizedPath::unchecked_new(ancestor);
742            if let Some(registration) = self.0.get(&WatchKey::exact(ancestor)) {
743                ids.extend_from_slice(&registration.watcher_ids);
744            }
745            if let Some(registration) = self.0.get(&WatchKey::folded(ancestor)) {
746                ids.extend_from_slice(&registration.watcher_ids);
747            }
748        }
749    }
750}
751
752struct WatcherState {
753    watchers: HashMap<WatcherRegistrationId, WatcherRegistrationState>,
754    native_path_registrations: WatchPaths,
755    poll_path_registrations: WatchPaths,
756    cooldown_until: Option<Instant>,
757    last_registration: WatcherRegistrationId,
758}
759
760impl WatcherState {
761    fn is_native_watch_limit_cooldown_active(&self) -> bool {
762        self.cooldown_until
763            .is_some_and(|cooldown_until| cooldown_until > Instant::now())
764    }
765
766    fn path_registrations(&mut self, mode: WatcherMode) -> &mut WatchPaths {
767        match mode {
768            WatcherMode::Native => &mut self.native_path_registrations,
769            WatcherMode::Poll => &mut self.poll_path_registrations,
770        }
771    }
772
773    fn remove_registration(
774        &mut self,
775        id: WatcherRegistrationId,
776    ) -> Option<(Arc<SanitizedPath>, WatcherMode)> {
777        let registration_state = self.watchers.remove(&id)?;
778        let path_registrations = self.path_registrations(registration_state.mode);
779        let path_state = path_registrations.get_mut(&registration_state.key)?;
780        path_state.watcher_ids.retain(|&existing| existing != id);
781        if !path_state.watcher_ids.is_empty() {
782            return None;
783        }
784
785        let was_actually_watched = path_state.has_os_watcher;
786        path_registrations.remove(&registration_state.key);
787
788        was_actually_watched.then_some((registration_state.path, registration_state.mode))
789    }
790}
791
792trait WatchBackend: Send {
793    fn watch(&mut self, path: &Path, mode: notify::RecursiveMode) -> notify::Result<()>;
794    fn unwatch(&mut self, path: &Path) -> notify::Result<()>;
795}
796
797impl<T: notify::Watcher + Send> WatchBackend for T {
798    fn watch(&mut self, path: &Path, mode: notify::RecursiveMode) -> notify::Result<()> {
799        notify::Watcher::watch(self, path, mode)
800    }
801
802    fn unwatch(&mut self, path: &Path) -> notify::Result<()> {
803        notify::Watcher::unwatch(self, path)
804    }
805}
806
807type DispatchEvent = (WatcherMode, Result<notify::Event, notify::Error>);
808
809pub struct GlobalWatcher {
810    state: Mutex<WatcherState>,
811
812    // DANGER: never keep state lock while holding watcher lock
813    // two mutexes because calling watcher.add triggers watcher.event, which needs watchers.
814    native_watcher: Mutex<Option<Box<dyn WatchBackend>>>,
815    poll_watcher: Mutex<Option<Box<dyn WatchBackend>>>,
816    event_tx: async_channel::Sender<DispatchEvent>,
817}
818
819impl GlobalWatcher {
820    #[must_use]
821    fn add(
822        &self,
823        path: Arc<std::path::Path>,
824        mode: WatcherMode,
825        case_insensitive: bool,
826        cb: impl Fn(&notify::Event) + Send + Sync + 'static,
827    ) -> anyhow::Result<Option<WatcherRegistrationId>> {
828        let path = SanitizedPath::from_arc(path);
829        let key = WatchKey::for_registration(&path, case_insensitive);
830        let mut state = self.state.lock();
831        let (path_already_covered, path_already_registered) = {
832            let registrations_for_mode = state.path_registrations(mode);
833            (
834                registrations_for_mode.covered_by_recursive_ancestor(&path, mode),
835                registrations_for_mode.contains(&key),
836            )
837        };
838
839        if !path_already_covered && !path_already_registered {
840            if mode == WatcherMode::Native && state.is_native_watch_limit_cooldown_active() {
841                return Ok(None);
842            }
843
844            drop(state);
845            match self.watch(path.as_path(), mode) {
846                Ok(()) => {}
847                Err(error) if mode == WatcherMode::Native && is_max_files_watch_error(&error) => {
848                    self.start_native_watch_limit_cooldown(path.as_path());
849                    return Ok(None);
850                }
851                Err(error) => return Err(error),
852            }
853            state = self.state.lock();
854        }
855
856        let id = state.last_registration;
857        state.last_registration = WatcherRegistrationId(id.0 + 1);
858
859        let registration_state = WatcherRegistrationState {
860            callback: Arc::new(cb),
861            key: key.clone(),
862            path,
863            mode,
864        };
865        state.watchers.insert(id, registration_state);
866        state
867            .path_registrations(mode)
868            .entry(key)
869            .and_modify(|registration| registration.watcher_ids.push(id))
870            .or_insert_with(|| PathRegistrationState {
871                watcher_ids: vec![id],
872                has_os_watcher: !path_already_covered,
873            });
874
875        Ok(Some(id))
876    }
877
878    fn enqueue(&self, mode: WatcherMode, event: Result<notify::Event, notify::Error>) {
879        if matches!(
880            event,
881            Ok(Event {
882                kind: EventKind::Access(_),
883                ..
884            })
885        ) {
886            return;
887        }
888
889        // A failed send only happens once the dispatch thread has shut down, at
890        // which point there's nothing left to dispatch to.
891        self.event_tx.try_send((mode, event)).ok();
892    }
893
894    fn dispatch(&self, mode: WatcherMode, event: Result<notify::Event, notify::Error>) {
895        let event = match event {
896            Ok(event) => event,
897            Err(error) => {
898                log::warn!("watcher error for {mode:?}: {error}");
899                return;
900            }
901        };
902
903        log::trace!("global handle event for {mode:?}: {event:?}");
904
905        let callbacks = {
906            let state = self.state.lock();
907            if event.need_rescan() {
908                let callbacks = state
909                    .watchers
910                    .values()
911                    .filter(|registration| registration.mode == mode)
912                    .map(|registration| registration.callback.clone())
913                    .collect::<Vec<_>>();
914                log::warn!(
915                    "filesystem watcher lost sync for {mode:?}; scheduling rescans for {} registrations",
916                    callbacks.len()
917                );
918                callbacks
919            } else {
920                let path_registrations = match mode {
921                    WatcherMode::Native => &state.native_path_registrations,
922                    WatcherMode::Poll => &state.poll_path_registrations,
923                };
924                let mut ids = Vec::new();
925                for path in &event.paths {
926                    let sanitized = SanitizedPath::new(path);
927                    path_registrations.watcher_ids_covering(sanitized, &mut ids);
928                }
929                ids.sort_unstable_by_key(|id| id.0);
930                ids.dedup();
931                ids.into_iter()
932                    .filter_map(|id| state.watchers.get(&id))
933                    .map(|registration| registration.callback.clone())
934                    .collect::<Vec<_>>()
935            }
936        };
937
938        for callback in callbacks {
939            callback(&event);
940        }
941    }
942
943    fn dispatch_batch(
944        &self,
945        first: DispatchEvent,
946        event_rx: &async_channel::Receiver<DispatchEvent>,
947    ) {
948        // A single backend overflow can enqueue many rescan markers. One rescan
949        // per mode covers the entire drained batch; ordinary events still run.
950        let mut native_rescan_dispatched = false;
951        let mut poll_rescan_dispatched = false;
952
953        for (mode, event) in
954            std::iter::once(first).chain(std::iter::from_fn(|| event_rx.try_recv().ok()))
955        {
956            let rescan_dispatched = match mode {
957                WatcherMode::Native => &mut native_rescan_dispatched,
958                WatcherMode::Poll => &mut poll_rescan_dispatched,
959            };
960            if event.as_ref().is_ok_and(notify::Event::need_rescan) {
961                if *rescan_dispatched {
962                    continue;
963                }
964                *rescan_dispatched = true;
965            }
966
967            self.dispatch(mode, event);
968        }
969    }
970
971    fn start_native_watch_limit_cooldown(&self, path: &Path) {
972        let mut state = self.state.lock();
973        let now = Instant::now();
974        let should_log = !state.is_native_watch_limit_cooldown_active();
975        state.cooldown_until = Some(now + *NATIVE_WATCH_LIMIT_COOLDOWN);
976        if should_log {
977            log::warn!(
978                "OS file watch limit reached while watching {path:?}; skipping new native file watcher registrations for {} seconds",
979                NATIVE_WATCH_LIMIT_COOLDOWN.as_secs()
980            );
981        }
982    }
983
984    pub fn remove(&self, id: WatcherRegistrationId) {
985        let mut state = self.state.lock();
986        let Some((path, mode)) = state.remove_registration(id) else {
987            return;
988        };
989        drop(state);
990        self.unwatch(path.as_path(), mode).log_err();
991    }
992
993    fn watch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> {
994        match mode {
995            WatcherMode::Native => {
996                self.ensure_native_watcher()?;
997                self.native_watcher
998                    .lock()
999                    .as_mut()
1000                    .expect("native watcher initialized")
1001                    .watch(
1002                        path,
1003                        if cfg!(any(target_os = "windows", target_os = "macos")) {
1004                            notify::RecursiveMode::Recursive
1005                        } else {
1006                            notify::RecursiveMode::NonRecursive
1007                        },
1008                    )?;
1009            }
1010            WatcherMode::Poll => {
1011                self.ensure_poll_watcher()?;
1012                self.poll_watcher
1013                    .lock()
1014                    .as_mut()
1015                    .expect("poll watcher initialized")
1016                    .watch(path, notify::RecursiveMode::Recursive)?;
1017            }
1018        }
1019
1020        Ok(())
1021    }
1022
1023    fn unwatch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> {
1024        let watcher = match mode {
1025            WatcherMode::Native => self
1026                .native_watcher
1027                .lock()
1028                .as_mut()
1029                .map(|watcher| watcher.unwatch(path)),
1030            WatcherMode::Poll => self
1031                .poll_watcher
1032                .lock()
1033                .as_mut()
1034                .map(|watcher| watcher.unwatch(path)),
1035        };
1036
1037        match watcher {
1038            // inotify auto-removes a watch when its directory is deleted, so a
1039            // later unwatch races that and fails with a benign error. Either way
1040            // the path is no longer watched, which is all we wanted.
1041            Some(Err(error)) if !matches!(error.kind, notify::ErrorKind::WatchNotFound) => {
1042                Err(error.into())
1043            }
1044            _ => Ok(()),
1045        }
1046    }
1047
1048    fn ensure_native_watcher(&self) -> anyhow::Result<()> {
1049        // The lock is held across creation: with a check-then-insert under two
1050        // separate lock acquisitions, concurrent callers could each create a
1051        // watcher and the loser's insert would silently drop the winner's
1052        // watcher along with every path registered on it.
1053        let mut native_watcher = self.native_watcher.lock();
1054        if native_watcher.is_none() {
1055            // CORE excludes Access events, which Zed discards anyway. Without this,
1056            // the default mask subscribes to inotify OPEN/CLOSE_* on Linux, so every
1057            // file read in a watched directory would queue events, increasing the
1058            // risk of queue overflows (and thus full rescans) under read-heavy
1059            // workloads like grep or language server indexing.
1060            let config = notify::Config::default().with_event_kinds(notify::EventKindMask::CORE);
1061            let watcher = <notify::RecommendedWatcher as notify::Watcher>::new(
1062                |event| global_watcher().enqueue(WatcherMode::Native, event),
1063                config,
1064            )?;
1065            *native_watcher = Some(Box::new(watcher));
1066        }
1067        Ok(())
1068    }
1069
1070    fn ensure_poll_watcher(&self) -> anyhow::Result<()> {
1071        let mut poll_watcher = self.poll_watcher.lock();
1072        if poll_watcher.is_none() {
1073            let config = notify::Config::default().with_poll_interval(*POLL_INTERVAL);
1074            let watcher = notify::PollWatcher::new(
1075                |event| global_watcher().enqueue(WatcherMode::Poll, event),
1076                config,
1077            )?;
1078            *poll_watcher = Some(Box::new(watcher));
1079        }
1080        Ok(())
1081    }
1082}
1083
1084fn is_max_files_watch_error(error: &anyhow::Error) -> bool {
1085    error
1086        .downcast_ref::<notify::Error>()
1087        .is_some_and(|error| matches!(&error.kind, notify::ErrorKind::MaxFilesWatch))
1088}
1089
1090static POLL_INTERVAL: LazyLock<Duration> = LazyLock::new(|| {
1091    let poll_ms: u64 = std::env::var("ZED_FILE_WATCHER_POLL_MS")
1092        .ok()
1093        .and_then(|value| value.parse().ok())
1094        .unwrap_or(2000)
1095        .clamp(500, 30000);
1096    Duration::from_millis(poll_ms)
1097});
1098
1099static NATIVE_WATCH_LIMIT_COOLDOWN: LazyLock<Duration> = LazyLock::new(|| {
1100    let cooldown_seconds: u64 = std::env::var("ZED_NATIVE_WATCH_LIMIT_COOLDOWN_SECONDS")
1101        .ok()
1102        .and_then(|value| value.parse().ok())
1103        .unwrap_or(5)
1104        .clamp(0, 300);
1105    Duration::from_secs(cooldown_seconds)
1106});
1107
1108pub fn poll_interval() -> Duration {
1109    *POLL_INTERVAL
1110}
1111
1112static FS_WATCHER_INSTANCE: OnceLock<GlobalWatcher> = OnceLock::new();
1113
1114fn global_watcher() -> &'static GlobalWatcher {
1115    FS_WATCHER_INSTANCE.get_or_init(|| {
1116        let (event_tx, event_rx) = async_channel::unbounded::<DispatchEvent>();
1117        std::thread::Builder::new()
1118            .name("fs-watcher-dispatch".to_owned())
1119            .spawn(move || {
1120                while let Ok(first) = event_rx.recv_blocking() {
1121                    global_watcher().dispatch_batch(first, &event_rx);
1122                }
1123            })
1124            .expect("failed to spawn fs watcher dispatch thread");
1125        GlobalWatcher {
1126            state: Mutex::new(WatcherState {
1127                watchers: Default::default(),
1128                native_path_registrations: Default::default(),
1129                poll_path_registrations: Default::default(),
1130                cooldown_until: None,
1131                last_registration: Default::default(),
1132            }),
1133            native_watcher: Mutex::new(None),
1134            poll_watcher: Mutex::new(None),
1135            event_tx,
1136        }
1137    })
1138}
1139
1140#[cfg(test)]
1141mod tests {
1142    use super::*;
1143    use std::{collections::HashSet, path::PathBuf};
1144
1145    fn rescan(path: &str) -> PathEvent {
1146        PathEvent {
1147            path: PathBuf::from(path),
1148            kind: Some(PathEventKind::Rescan),
1149        }
1150    }
1151
1152    fn changed(path: &str) -> PathEvent {
1153        PathEvent {
1154            path: PathBuf::from(path),
1155            kind: Some(PathEventKind::Changed),
1156        }
1157    }
1158
1159    #[derive(Default)]
1160    struct FakeWatchBackend {
1161        watched_paths: HashSet<PathBuf>,
1162        watch_calls: Vec<PathBuf>,
1163        unwatch_calls: Vec<PathBuf>,
1164        fail_with_watch_limit: bool,
1165    }
1166
1167    struct SharedFakeWatchBackend(Arc<Mutex<FakeWatchBackend>>);
1168
1169    impl WatchBackend for SharedFakeWatchBackend {
1170        fn watch(&mut self, path: &Path, _mode: notify::RecursiveMode) -> notify::Result<()> {
1171            let path = path.to_path_buf();
1172            let mut backend = self.0.lock();
1173            backend.watch_calls.push(path.clone());
1174            if backend.fail_with_watch_limit {
1175                return Err(notify::Error::new(notify::ErrorKind::MaxFilesWatch));
1176            }
1177            backend.watched_paths.insert(path);
1178            Ok(())
1179        }
1180
1181        fn unwatch(&mut self, path: &Path) -> notify::Result<()> {
1182            let path = path.to_path_buf();
1183            let mut backend = self.0.lock();
1184            backend.unwatch_calls.push(path.clone());
1185            if backend.watched_paths.remove(&path) {
1186                Ok(())
1187            } else {
1188                Err(notify::Error::generic("path was not watched"))
1189            }
1190        }
1191    }
1192
1193    fn test_watcher(poll_watcher: Arc<Mutex<FakeWatchBackend>>) -> GlobalWatcher {
1194        test_watcher_with_backends(None, Some(poll_watcher))
1195    }
1196
1197    fn test_watcher_with_backends(
1198        native_watcher: Option<Arc<Mutex<FakeWatchBackend>>>,
1199        poll_watcher: Option<Arc<Mutex<FakeWatchBackend>>>,
1200    ) -> GlobalWatcher {
1201        // Tests call `handle_event` directly to exercise dispatch synchronously,
1202        // rather than going through the OS watcher callbacks, so nothing is ever
1203        // sent on this channel; the receiver can just be dropped.
1204        let (event_tx, _event_rx) = async_channel::unbounded();
1205        GlobalWatcher {
1206            state: Mutex::new(WatcherState {
1207                watchers: Default::default(),
1208                native_path_registrations: Default::default(),
1209                poll_path_registrations: Default::default(),
1210                cooldown_until: None,
1211                last_registration: Default::default(),
1212            }),
1213            native_watcher: Mutex::new(
1214                native_watcher.map(|watcher| {
1215                    Box::new(SharedFakeWatchBackend(watcher)) as Box<dyn WatchBackend>
1216                }),
1217            ),
1218            poll_watcher: Mutex::new(
1219                poll_watcher.map(|watcher| {
1220                    Box::new(SharedFakeWatchBackend(watcher)) as Box<dyn WatchBackend>
1221                }),
1222            ),
1223            event_tx,
1224        }
1225    }
1226
1227    struct TestCase {
1228        name: &'static str,
1229        pending_paths: Vec<PathEvent>,
1230        path_events: Vec<PathEvent>,
1231        expected_pending_paths: Vec<PathEvent>,
1232        expected_path_events: Vec<PathEvent>,
1233    }
1234
1235    #[test]
1236    fn covered_child_registration_is_not_unwatched_after_parent_is_removed() {
1237        let backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
1238        let watcher = test_watcher(backend.clone());
1239        let parent = Arc::<Path>::from(Path::new("/repo"));
1240        let child = Arc::<Path>::from(Path::new("/repo/foo.csproj"));
1241
1242        let parent_registration = watcher
1243            .add(parent.as_ref().into(), WatcherMode::Poll, false, |_| {})
1244            .expect("add parent watch")
1245            .expect("parent watch registered");
1246        let child_registration = watcher
1247            .add(child.as_ref().into(), WatcherMode::Poll, false, |_| {})
1248            .expect("add covered child watch")
1249            .expect("child watch registered");
1250
1251        watcher.remove(parent_registration);
1252        watcher.remove(child_registration);
1253
1254        let backend = backend.lock();
1255        assert_eq!(backend.watch_calls, &[parent.to_path_buf()]);
1256        assert_eq!(backend.unwatch_calls, &[parent.to_path_buf()]);
1257    }
1258
1259    #[gpui::test]
1260    async fn pending_path_is_registered_once_created(cx: &mut gpui::TestAppContext) {
1261        let temp_dir = tempfile::tempdir().expect("create temp dir");
1262        let path = temp_dir.path().join("file.txt");
1263
1264        let (tx, rx) = async_channel::unbounded();
1265        let pending_path_events: Arc<Mutex<Vec<PathEvent>>> = Default::default();
1266        let watcher = FsWatcher::new(cx.executor(), tx, pending_path_events.clone());
1267
1268        watcher
1269            .add(&path)
1270            .expect("add path that does not exist yet");
1271        assert!(
1272            watcher
1273                .pending_registrations
1274                .lock()
1275                .contains_key(path.as_path())
1276        );
1277        assert!(watcher.registrations.lock().is_empty());
1278
1279        std::fs::write(&path, b"contents").expect("create path");
1280
1281        // poll_path_until_created stats the path on smol's blocking pool, which
1282        // the deterministic executor cannot drive; park until the poll task
1283        // signals the event channel.
1284        cx.executor().allow_parking();
1285        cx.executor().advance_clock(poll_interval());
1286        rx.recv().await.expect("receive watcher event");
1287
1288        assert!(
1289            !watcher
1290                .pending_registrations
1291                .lock()
1292                .contains_key(path.as_path())
1293        );
1294        let case_insensitive = case_insensitive_path(&path);
1295        let key = WatchKey::for_registration(SanitizedPath::new(&path), case_insensitive);
1296        assert!(watcher.registrations.lock().contains_key(&key));
1297
1298        // poll_path_until_created also enqueues a Rescan for the same path, but
1299        // enqueue_path_events -> util::extend_sorted dedups by path, so only Created survives.
1300        assert_eq!(
1301            pending_path_events.lock().clone(),
1302            vec![PathEvent {
1303                path: path.clone(),
1304                kind: Some(PathEventKind::Created),
1305            }]
1306        );
1307    }
1308
1309    #[test]
1310    fn native_watch_limit_cools_down_subsequent_native_registrations() {
1311        let native_backend = Arc::new(Mutex::new(FakeWatchBackend {
1312            fail_with_watch_limit: true,
1313            ..Default::default()
1314        }));
1315        let poll_backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
1316        let watcher = test_watcher_with_backends(Some(native_backend.clone()), Some(poll_backend));
1317        let first_path = Arc::<Path>::from(Path::new("/repo/first"));
1318        let second_path = Arc::<Path>::from(Path::new("/repo/second"));
1319
1320        let first_registration = watcher
1321            .add(first_path.clone(), WatcherMode::Native, false, |_| {})
1322            .expect("native watch limit is handled");
1323        let second_registration = watcher
1324            .add(second_path, WatcherMode::Native, false, |_| {})
1325            .expect("native watch limit backoff is handled");
1326
1327        assert!(first_registration.is_none());
1328        assert!(second_registration.is_none());
1329
1330        let native_backend = native_backend.lock();
1331        assert_eq!(native_backend.watch_calls, &[first_path.to_path_buf()]);
1332    }
1333
1334    fn modify_event(path: &str) -> notify::Event {
1335        notify::Event {
1336            paths: vec![PathBuf::from(path)],
1337            ..notify::Event::new(EventKind::Modify(notify::event::ModifyKind::Any))
1338        }
1339    }
1340
1341    fn recording_watcher() -> (GlobalWatcher, Arc<Mutex<Vec<String>>>) {
1342        let backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
1343        let watcher = test_watcher_with_backends(Some(backend), None);
1344        let fired = Arc::new(Mutex::new(Vec::new()));
1345        for dir in ["/repo/a", "/repo/a/nested", "/repo/b"] {
1346            let fired = fired.clone();
1347            let label = dir.to_owned();
1348            watcher
1349                .add(
1350                    Arc::<Path>::from(Path::new(dir)),
1351                    WatcherMode::Native,
1352                    false,
1353                    move |_| {
1354                        fired.lock().push(label.clone());
1355                    },
1356                )
1357                .expect("add watch")
1358                .expect("watch registered");
1359        }
1360        (watcher, fired)
1361    }
1362
1363    #[test]
1364    fn event_dispatches_only_to_registrations_covering_its_path() {
1365        let (watcher, fired) = recording_watcher();
1366
1367        watcher.dispatch(WatcherMode::Native, Ok(modify_event("/repo/a/file.txt")));
1368
1369        // Only the directory containing the file resolves; siblings stay untouched.
1370        assert_eq!(*fired.lock(), vec!["/repo/a".to_owned()]);
1371    }
1372
1373    #[test]
1374    fn event_dispatches_to_every_ancestor_registration() {
1375        let (watcher, fired) = recording_watcher();
1376
1377        watcher.dispatch(
1378            WatcherMode::Native,
1379            Ok(modify_event("/repo/a/nested/file.txt")),
1380        );
1381
1382        // Both the directory containing the file and the ancestor watching it
1383        // resync, each exactly once, matching the previous broadcast-and-filter
1384        // behavior.
1385        let mut got = fired.lock().clone();
1386        got.sort();
1387        assert_eq!(got, vec!["/repo/a".to_owned(), "/repo/a/nested".to_owned()]);
1388    }
1389
1390    fn fired_count() -> (
1391        Arc<Mutex<usize>>,
1392        impl Fn(&notify::Event) + Send + Sync + 'static,
1393    ) {
1394        let fired = Arc::new(Mutex::new(0usize));
1395        let cb = {
1396            let fired = fired.clone();
1397            move |_: &notify::Event| *fired.lock() += 1
1398        };
1399        (fired, cb)
1400    }
1401
1402    #[test]
1403    fn watch_key_folds_case_but_keeps_exact_distinct() {
1404        let mixed = SanitizedPath::new(Path::new("/Repo/Proj"));
1405        let lower = SanitizedPath::new(Path::new("/repo/proj"));
1406
1407        // Folded keys collide regardless of casing; exact keys do not.
1408        assert_eq!(WatchKey::folded(mixed), WatchKey::folded(lower));
1409        assert_ne!(WatchKey::exact(mixed), WatchKey::exact(lower));
1410        // Exact and folded live in different key spaces even for the same path.
1411        assert_ne!(WatchKey::exact(mixed), WatchKey::folded(mixed));
1412    }
1413
1414    #[cfg(target_os = "macos")]
1415    #[test]
1416    fn watch_key_folds_unicode_normalization_on_macos() {
1417        // "Café" precomposed (NFC) vs decomposed (NFD) are different byte
1418        // sequences but the same directory on a normalization-insensitive volume.
1419        let nfc = Path::new("/repo/Caf\u{00e9}");
1420        let nfd = Path::new("/repo/Cafe\u{0301}");
1421        assert_ne!(nfc, nfd);
1422        assert_eq!(
1423            WatchKey::folded(SanitizedPath::new(nfc)),
1424            WatchKey::folded(SanitizedPath::new(nfd)),
1425        );
1426    }
1427
1428    #[test]
1429    fn case_insensitive_registration_matches_differently_cased_event() {
1430        let (fired, cb) = fired_count();
1431        let watcher = test_watcher_with_backends(Some(Default::default()), None);
1432        watcher
1433            .add(
1434                Path::new("/Repo/Project").into(),
1435                WatcherMode::Native,
1436                true,
1437                cb,
1438            )
1439            .expect("add")
1440            .expect("registered");
1441
1442        // Event arrives lowercased (as TSGO/macOS may report it).
1443        watcher.dispatch(
1444            WatcherMode::Native,
1445            Ok(modify_event("/repo/project/file.txt")),
1446        );
1447        assert_eq!(*fired.lock(), 1);
1448    }
1449
1450    #[test]
1451    fn case_insensitive_registration_survives_case_only_rename() {
1452        let (fired, cb) = fired_count();
1453        let watcher = test_watcher_with_backends(Some(Default::default()), None);
1454        watcher
1455            .add(
1456                Path::new("/Repo/Proj").into(),
1457                WatcherMode::Native,
1458                true,
1459                cb,
1460            )
1461            .expect("add")
1462            .expect("registered");
1463
1464        // The watched directory was renamed to a different casing; events now
1465        // arrive under the new spelling.
1466        watcher.dispatch(WatcherMode::Native, Ok(modify_event("/Repo/PROJ/file.txt")));
1467        assert_eq!(*fired.lock(), 1);
1468    }
1469
1470    #[test]
1471    fn case_sensitive_registration_ignores_differently_cased_event() {
1472        let (fired, cb) = fired_count();
1473        let watcher = test_watcher_with_backends(Some(Default::default()), None);
1474        watcher
1475            .add(
1476                Path::new("/Repo/proj").into(),
1477                WatcherMode::Native,
1478                false,
1479                cb,
1480            )
1481            .expect("add")
1482            .expect("registered");
1483
1484        // On a case-sensitive volume these are genuinely different directories.
1485        watcher.dispatch(WatcherMode::Native, Ok(modify_event("/Repo/PROJ/file.txt")));
1486        assert_eq!(*fired.lock(), 0);
1487    }
1488
1489    #[test]
1490    fn differently_cased_adds_dedupe_on_case_insensitive_volume() {
1491        let backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
1492        let watcher = test_watcher_with_backends(Some(backend.clone()), None);
1493        watcher
1494            .add(
1495                Path::new("/Repo/Proj").into(),
1496                WatcherMode::Native,
1497                true,
1498                |_| {},
1499            )
1500            .expect("add")
1501            .expect("registered");
1502        watcher
1503            .add(
1504                Path::new("/repo/proj").into(),
1505                WatcherMode::Native,
1506                true,
1507                |_| {},
1508            )
1509            .expect("add")
1510            .expect("registered");
1511
1512        // The second, differently-cased spelling reuses the same OS watch.
1513        assert_eq!(backend.lock().watch_calls.len(), 1);
1514    }
1515
1516    #[test]
1517    fn recursive_parent_covers_differently_cased_child() {
1518        let backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
1519        let watcher = test_watcher(backend.clone());
1520        watcher
1521            .add(Path::new("/Repo").into(), WatcherMode::Poll, true, |_| {})
1522            .expect("add")
1523            .expect("registered");
1524        watcher
1525            .add(
1526                Path::new("/repo/child").into(),
1527                WatcherMode::Poll,
1528                true,
1529                |_| {},
1530            )
1531            .expect("add");
1532
1533        // The child is covered by the recursive parent despite the case mismatch.
1534        assert_eq!(backend.lock().watch_calls, vec![PathBuf::from("/Repo")]);
1535    }
1536
1537    #[test]
1538    fn rescan_event_broadcasts_to_all_registrations_of_the_same_mode() {
1539        let (watcher, fired) = recording_watcher();
1540
1541        let rescan = notify::Event::new(EventKind::Other).set_flag(notify::event::Flag::Rescan);
1542        watcher.dispatch(WatcherMode::Native, Ok(rescan));
1543
1544        // A pathless rescan may have missed events anywhere, so every registration
1545        // of that mode resyncs, regardless of which directory it watches.
1546        let mut got = fired.lock().clone();
1547        got.sort();
1548        assert_eq!(
1549            got,
1550            vec![
1551                "/repo/a".to_owned(),
1552                "/repo/a/nested".to_owned(),
1553                "/repo/b".to_owned(),
1554            ]
1555        );
1556    }
1557
1558    #[test]
1559    fn queued_rescans_are_coalesced_without_dropping_normal_events() {
1560        let (watcher, fired) = recording_watcher();
1561        let (event_tx, event_rx) = async_channel::unbounded();
1562        let rescan = || notify::Event::new(EventKind::Other).set_flag(notify::event::Flag::Rescan);
1563
1564        event_tx
1565            .try_send((WatcherMode::Native, Ok(rescan())))
1566            .unwrap();
1567        event_tx
1568            .try_send((WatcherMode::Native, Ok(modify_event("/repo/a/file.txt"))))
1569            .unwrap();
1570        watcher.dispatch_batch((WatcherMode::Native, Ok(rescan())), &event_rx);
1571
1572        let mut got = fired.lock().clone();
1573        got.sort();
1574        assert_eq!(
1575            got,
1576            vec![
1577                "/repo/a".to_owned(),
1578                "/repo/a".to_owned(),
1579                "/repo/a/nested".to_owned(),
1580                "/repo/b".to_owned(),
1581            ]
1582        );
1583    }
1584
1585    #[test]
1586    #[cfg(target_os = "windows")]
1587    fn event_dispatches_when_reported_path_has_verbatim_prefix() {
1588        // `notify` (and Windows APIs generally) can report a changed path using the
1589        // verbatim `\\?\` long-path form even when the directory was registered
1590        // without it.
1591        let backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
1592        let watcher = test_watcher_with_backends(Some(backend), None);
1593        let fired = Arc::new(Mutex::new(Vec::new()));
1594        {
1595            let fired = fired.clone();
1596            watcher
1597                .add(
1598                    Arc::<Path>::from(Path::new("C:\\repo\\src")),
1599                    WatcherMode::Native,
1600                    false,
1601                    move |_| fired.lock().push(()),
1602                )
1603                .expect("add watch")
1604                .expect("watch registered");
1605        }
1606
1607        watcher.dispatch(
1608            WatcherMode::Native,
1609            Ok(modify_event("\\\\?\\C:\\repo\\src\\main.rs")),
1610        );
1611
1612        assert_eq!(fired.lock().len(), 1);
1613    }
1614
1615    #[test]
1616    fn test_coalesce_pending_rescans() {
1617        let test_cases = [
1618            TestCase {
1619                name: "coalesces descendant rescans under pending ancestor",
1620                pending_paths: vec![rescan("/root")],
1621                path_events: vec![rescan("/root/child"), rescan("/root/child/grandchild")],
1622                expected_pending_paths: vec![rescan("/root")],
1623                expected_path_events: vec![],
1624            },
1625            TestCase {
1626                name: "new ancestor rescan replaces pending descendant rescans",
1627                pending_paths: vec![
1628                    changed("/other"),
1629                    rescan("/root/child"),
1630                    rescan("/root/child/grandchild"),
1631                ],
1632                path_events: vec![rescan("/root")],
1633                expected_pending_paths: vec![changed("/other")],
1634                expected_path_events: vec![rescan("/root")],
1635            },
1636            TestCase {
1637                name: "same path rescan replaces pending non-rescan event",
1638                pending_paths: vec![changed("/root")],
1639                path_events: vec![rescan("/root")],
1640                expected_pending_paths: vec![],
1641                expected_path_events: vec![rescan("/root")],
1642            },
1643            TestCase {
1644                name: "unrelated rescans are preserved",
1645                pending_paths: vec![rescan("/root-a")],
1646                path_events: vec![rescan("/root-b")],
1647                expected_pending_paths: vec![rescan("/root-a")],
1648                expected_path_events: vec![rescan("/root-b")],
1649            },
1650            TestCase {
1651                name: "batch ancestor rescan replaces descendant rescan",
1652                pending_paths: vec![],
1653                path_events: vec![rescan("/root/child"), rescan("/root")],
1654                expected_pending_paths: vec![],
1655                expected_path_events: vec![rescan("/root")],
1656            },
1657        ];
1658
1659        for test_case in test_cases {
1660            let mut pending_paths = test_case.pending_paths;
1661            let mut path_events = test_case.path_events;
1662
1663            coalesce_pending_rescans(&mut pending_paths, &mut path_events);
1664
1665            assert_eq!(
1666                pending_paths, test_case.expected_pending_paths,
1667                "pending_paths mismatch for case: {}",
1668                test_case.name
1669            );
1670            assert_eq!(
1671                path_events, test_case.expected_path_events,
1672                "path_events mismatch for case: {}",
1673                test_case.name
1674            );
1675        }
1676    }
1677}
1678
1679pub fn global<T>(f: impl FnOnce(&GlobalWatcher) -> T) -> anyhow::Result<T> {
1680    let global_watcher = global_watcher();
1681    global_watcher.ensure_native_watcher()?;
1682    Ok(f(global_watcher))
1683}
1684