fix: keep macOS watchers within FSEvents path limits

Author: masonjames

src/watcher/unified.rs

@@ -100,12 +100,10 @@ impl UnifiedWatcher {
             );
         }
 
-        // Watch all directories
-        for dir in new_dirs {
-            self.watch_directory(&dir)?;
-        }
-
+        // FSEvents watches roots recursively; register those first so the
+        // per-directory batch can skip paths they already cover.
         self.register_handler_roots().await;
+        self.watch_directories(&new_dirs)?;
 
         // Subscribe to broadcaster for IndexReloaded events
         let mut broadcast_rx = self.broadcaster.subscribe();
@@ -180,36 +178,53 @@ impl UnifiedWatcher {
         }
     }
 
-    /// Watch a directory for changes.
-    fn watch_directory(&mut self, dir: &PathBuf) -> Result<(), WatchError> {
-        let watch_path = if dir.is_absolute() {
-            dir.clone()
+    /// Register a batch without restarting the platform event stream per path.
+    fn watch_directories(&mut self, dirs: &[PathBuf]) -> Result<(), WatchError> {
+        let watch_paths: Vec<_> = dirs
+            .iter()
+            .map(|dir| {
+                if dir.is_absolute() {
+                    dir.clone()
+                } else {
+                    self.workspace_root.join(dir)
+                }
+            })
+            .filter(|path| {
+                !cfg!(target_os = "macos")
+                    || !self.handler_roots.iter().any(|root| path.starts_with(root))
+            })
+            .collect();
+        if watch_paths.is_empty() {
+            return Ok(());
+        }
+        // FSEvents has a finite path list. Recursive roots avoid one native
+        // watch per source directory; handlers still enforce ignore rules.
+        let mode = if cfg!(target_os = "macos") {
+            RecursiveMode::Recursive
         } else {
-            self.workspace_root.join(dir)
+            RecursiveMode::NonRecursive
         };
-
-        match self
-            ._watcher
-            .watch(&watch_path, RecursiveMode::NonRecursive)
-        {
-            Ok(_) => {
-                crate::debug_event!(
-                    "watcher",
-                    "watching",
-                    "{}",
-                    crate::parsing::paths::render_absolute_path(&watch_path).display()
-                );
-                Ok(())
-            }
-            Err(e) => {
-                tracing::warn!(
-                    "[watcher] failed to watch {}: {e}",
-                    crate::parsing::paths::render_absolute_path(&watch_path).display()
-                );
-                // Continue - don't fail completely
-                Ok(())
+        let mut paths = self._watcher.paths_mut();
+        for watch_path in watch_paths {
+            match paths.add(&watch_path, mode) {
+                Ok(_) => {
+                    crate::debug_event!(
+                        "watcher",
+                        "watching",
+                        "{}",
+                        crate::parsing::paths::render_absolute_path(&watch_path).display()
+                    );
+                }
+                Err(e) => {
+                    tracing::warn!(
+                        "[watcher] failed to watch {}: {e}",
+                        crate::parsing::paths::render_absolute_path(&watch_path).display()
+                    );
+                }
             }
         }
+        paths.commit()?;
+        Ok(())
     }
 
     /// Handle an incoming file event.
@@ -305,12 +320,16 @@ impl UnifiedWatcher {
             }
             roots.extend(handler_roots);
         }
-        for root in &roots {
-            if self.registry.add_watch_dir(root.clone()) {
-                if let Err(e) = self.watch_directory(root) {
-                    tracing::warn!("[watcher] failed to watch root: {e}");
-                }
-            }
+        let new_roots: Vec<_> = roots
+            .iter()
+            .filter(|root| {
+                let new_dir = self.registry.add_watch_dir((*root).clone());
+                new_dir || (cfg!(target_os = "macos") && !self.handler_roots.contains(root))
+            })
+            .cloned()
+            .collect();
+        if let Err(e) = self.watch_directories(&new_roots) {
+            tracing::warn!("[watcher] failed to watch roots: {e}");
         }
         self.handler_roots = roots;
         self.batch_sync_roots = sync_roots;
@@ -333,12 +352,12 @@ impl UnifiedWatcher {
             )
         };
 
-        for dir in dirs {
-            if self.registry.add_watch_dir(dir.clone()) {
-                if let Err(e) = self.watch_directory(&dir) {
-                    tracing::warn!("[watcher] failed to watch created dir: {e}");
-                }
-            }
+        let new_dirs: Vec<_> = dirs
+            .into_iter()
+            .filter(|dir| self.registry.add_watch_dir(dir.clone()))
+            .collect();
+        if let Err(e) = self.watch_directories(&new_dirs) {
+            tracing::warn!("[watcher] failed to watch created directories: {e}");
         }
         if !files.is_empty() {
             crate::log_event!(
@@ -716,10 +735,8 @@ impl UnifiedWatcher {
             .collect();
 
         // Watch any new directories
-        for dir in dirs_to_watch {
-            if let Err(e) = self.watch_directory(&dir) {
-                tracing::warn!("[watcher] failed to watch new directory: {e}");
-            }
+        if let Err(e) = self.watch_directories(&dirs_to_watch) {
+            tracing::warn!("[watcher] failed to watch new directories: {e}");
         }
 
         // Config reload can add or drop roots; re-register them.

tests/watcher_startup.rs

@@ -0,0 +1,113 @@
+#![cfg(target_os = "macos")]
+
+use std::{
+    fs,
+    process::{Command, Stdio},
+    thread,
+    time::{Duration, Instant},
+};
+use tantivy::{Index, Term, collector::Count, query::TermQuery, schema::IndexRecordOption};
+
+#[test]
+#[ignore = "creates 5,000 directories; run explicitly to check FSEvents scaling and delivery"]
+fn watches_large_tree_and_preserves_ignore_rules() {
+    let dir = tempfile::tempdir().unwrap();
+    let root = dir.path().canonicalize().unwrap();
+    fs::create_dir(root.join(".codanna")).unwrap();
+    fs::write(root.join(".codanna/settings.toml"), format!(
+        "workspace_root = {}\n[indexing]\nindexed_paths = [\"src\"]\n[semantic_search]\nenabled = false\n",
+        serde_json::to_string(&root).unwrap()
+    )).unwrap();
+    for i in 0..5000 {
+        let path = root.join(format!("src/d{i}"));
+        fs::create_dir_all(&path).unwrap();
+        fs::write(
+            path.join("seed.py"),
+            format!("def seed_{i}():\n    return {i}\n"),
+        )
+        .unwrap();
+    }
+    // Direct root files must not suppress the native root watch just because
+    // their parent is already in the logical directory registry.
+    fs::write(root.join("src/seed.py"), "def root_seed():\n    return 0\n").unwrap();
+    fs::write(root.join("src/.codannaignore"), "ignored/\n").unwrap();
+    let binary = env!("CARGO_BIN_EXE_codanna");
+    assert!(
+        Command::new(binary)
+            .current_dir(&root)
+            .args(["index", "--no-progress"])
+            .stdout(Stdio::null())
+            .status()
+            .unwrap()
+            .success()
+    );
+    let index = Index::open_in_dir(root.join(".codanna/index/tantivy")).unwrap();
+    let reader = index.reader().unwrap();
+    let name_field = index.schema().get_field("name").unwrap();
+    let contains = |name: &str| {
+        reader.reload().unwrap();
+        reader
+            .searcher()
+            .search(
+                &TermQuery::new(
+                    Term::from_field_text(name_field, name),
+                    IndexRecordOption::Basic,
+                ),
+                &Count,
+            )
+            .unwrap()
+            > 0
+    };
+    let wait_for = |condition: &dyn Fn() -> bool| {
+        let start = Instant::now();
+        while start.elapsed() < Duration::from_secs(15) {
+            if condition() {
+                return true;
+            }
+            thread::sleep(Duration::from_millis(100));
+        }
+        false
+    };
+    let log = root.join("watcher.log");
+    let mut child = Command::new(binary)
+        .current_dir(&root)
+        .args(["serve", "--watch"])
+        .env("RUST_LOG", "codanna::watcher::unified=info")
+        .stdin(Stdio::piped())
+        .stdout(Stdio::null())
+        .stderr(fs::File::create(&log).unwrap())
+        .spawn()
+        .unwrap();
+    let start = Instant::now();
+    let ready = wait_for(&|| {
+        fs::read_to_string(&log)
+            .unwrap()
+            .contains("[watcher] started")
+    });
+    let ready_elapsed = start.elapsed();
+    fs::create_dir(root.join("src/ignored")).unwrap();
+    fs::write(
+        root.join("src/ignored/skip.py"),
+        "def ignored_probe():\n    return 0\n",
+    )
+    .unwrap();
+    fs::create_dir(root.join("src/new_subtree")).unwrap();
+    let source = root.join("src/new_subtree/created.py");
+    fs::write(&source, "def created_probe():\n    return 1\n").unwrap();
+    let created = wait_for(&|| contains("created_probe"));
+    fs::write(&source, "def edited_probe():\n    return 2\n").unwrap();
+    let edited = wait_for(&|| contains("edited_probe") && !contains("created_probe"));
+    fs::remove_file(&source).unwrap();
+    let deleted = wait_for(&|| !contains("edited_probe"));
+    let ignored = contains("ignored_probe");
+    let _ = child.kill();
+    child.wait().unwrap();
+    assert!(
+        ready && created && edited && deleted && !ignored,
+        "ready={ready} created={created} edited={edited} deleted={deleted} ignored={ignored}: {}",
+        fs::read_to_string(log).unwrap()
+    );
+    println!(
+        "5,000-directory readiness: {ready_elapsed:?}; create/edit/delete and ignore checks passed"
+    );
+}