fix(watcher): make watch-triggered syncs incremental instead of full-tree

On large repos, notify-debouncer-full periodically rescans the whole
watched tree (its rename-detection/overflow recovery mechanism),
emitting synthetic events for most files even with zero real edits.
Since the previous fix only filtered out .codegraph/.git/.gitignore
paths, every rescan still triggered a full repo walk + rayon parse of
the entire tree, sustaining ~1000% CPU and steadily growing RSS
indefinitely - confirmed by cloning the reporter's repo (tnl) and
reproducing the exact symptom.

Orchestrator::sync_paths() now processes only the specific paths
reported by the watcher (parsing/upserting changed files, deleting
removed ones) instead of re-walking and re-parsing every file in the
repo on every debounce tick. The watcher also coalesces any batches
queued during processing to avoid back-to-back passes. Verified on
the same repo: CPU/RSS stay flat (0% CPU, ~10MB RSS) over 2 minutes
with no edits, and a real file edit still triggers a correct 1-file
incremental sync.

Author: Cleboost

crates/codegraph-extract/src/orchestrator.rs

@@ -1,5 +1,5 @@
 use crate::{walker, ExtractResult, Extractor};
-use camino::Utf8Path;
+use camino::{Utf8Path, Utf8PathBuf};
 use codegraph_core::Result;
 use codegraph_db::{Db, EdgeDraft, FileRow, NodeDraft};
 use codegraph_resolve::{PendingCallRow, Resolver};
@@ -41,7 +41,40 @@ impl Orchestrator {
             .par_iter()
             .filter_map(|fm| parse_one(fm).ok().flatten())
             .collect();
+        self.apply(db, parsed)
+    }
+
+    /// Sync only the given paths instead of walking the whole tree. Used by the
+    /// watcher so that a burst of filesystem events costs O(changed files),
+    /// not O(repo size).
+    pub fn sync_paths(&self, db: &Db, paths: &[Utf8PathBuf]) -> Result<ExtractStats> {
+        let ext_map = walker::build_ext_map(&self.extractors);
+        let mut matches = Vec::new();
+        for p in paths {
+            if !p.as_std_path().is_file() {
+                // Deleted (or not a regular file): drop it from the index if present.
+                if let Ok(Some(existing)) = db.file_by_path(p.as_str()) {
+                    if let Some(eid) = existing.id {
+                        db.delete_file_cascade(eid)?;
+                    }
+                }
+                continue;
+            }
+            if let Some(extractor) = walker::match_extractor(p, &ext_map) {
+                matches.push(walker::FileMatch {
+                    path: p.clone(),
+                    extractor,
+                });
+            }
+        }
+        let parsed: Vec<_> = matches
+            .par_iter()
+            .filter_map(|fm| parse_one(fm).ok().flatten())
+            .collect();
+        self.apply(db, parsed)
+    }
 
+    fn apply(&self, db: &Db, parsed: Vec<Parsed>) -> Result<ExtractStats> {
         let mut stats = ExtractStats::default();
         let mut all_pending: Vec<PendingCallRow> = Vec::new();
         for Parsed { row, result } in parsed {

crates/codegraph-extract/src/walker.rs

@@ -9,13 +9,28 @@ pub struct FileMatch {
     pub extractor: Arc<dyn Extractor>,
 }
 
-pub fn walk(root: &Utf8Path, extractors: &[Arc<dyn Extractor>]) -> Vec<FileMatch> {
-    let mut ext_map: HashMap<&'static str, Arc<dyn Extractor>> = HashMap::new();
+pub type ExtMap = HashMap<&'static str, Arc<dyn Extractor>>;
+
+pub fn build_ext_map(extractors: &[Arc<dyn Extractor>]) -> ExtMap {
+    let mut ext_map: ExtMap = HashMap::new();
     for ex in extractors {
         for e in ex.extensions() {
             ext_map.insert(*e, ex.clone());
         }
     }
+    ext_map
+}
+
+/// Match a single path against the extractor registry, without walking the tree.
+/// Used for incremental (watcher-driven) syncs where the caller already knows
+/// which paths changed.
+pub fn match_extractor(path: &Utf8Path, ext_map: &ExtMap) -> Option<Arc<dyn Extractor>> {
+    let ext = path.extension()?;
+    ext_map.get(ext).cloned()
+}
+
+pub fn walk(root: &Utf8Path, extractors: &[Arc<dyn Extractor>]) -> Vec<FileMatch> {
+    let ext_map = build_ext_map(extractors);
 
     let mut out = Vec::new();
     let walker = WalkBuilder::new(root)

crates/codegraph/src/watcher.rs

@@ -2,9 +2,10 @@ use anyhow::Result;
 use camino::Utf8PathBuf;
 use codegraph_db::Db;
 use codegraph_extract::Orchestrator;
-use ignore::gitignore::GitignoreBuilder;
+use ignore::gitignore::{Gitignore, GitignoreBuilder};
 use notify::RecursiveMode;
 use notify_debouncer_full::{new_debouncer, DebouncedEvent};
+use std::collections::BTreeSet;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -42,19 +43,19 @@ fn run(root: Utf8PathBuf, db: Arc<Db>) -> Result<()> {
 
     let orch = Orchestrator::with_registry();
     while let Ok(events) = rx.recv() {
-        let relevant = events.iter().any(|event| {
-            event.paths.iter().any(|p| {
-                let under_ignored_dir = ignored_dirs.iter().any(|dir| p.starts_with(dir.as_std_path()));
-                if under_ignored_dir {
-                    return false;
-                }
-                !gitignore.matched(p, p.is_dir()).is_ignore()
-            })
-        });
-        if !relevant {
+        let mut batch = events;
+        // Coalesce any batches that arrive while we're about to process one -
+        // avoids back-to-back sync passes when the debouncer fires repeatedly
+        // in quick succession (e.g. during a large rescan).
+        while let Ok(more) = rx.try_recv() {
+            batch.extend(more);
+        }
+
+        let paths = relevant_paths(&batch, &root, &ignored_dirs, &gitignore);
+        if paths.is_empty() {
             continue;
         }
-        match orch.sync(&root, &db) {
+        match orch.sync_paths(&db, &paths) {
             Ok(s) if s.files > 0 => {
                 tracing::info!("watch sync: {} files, {} edges", s.files, s.edges)
             }
@@ -64,3 +65,32 @@ fn run(root: Utf8PathBuf, db: Arc<Db>) -> Result<()> {
     }
     Ok(())
 }
+
+fn relevant_paths(
+    events: &[DebouncedEvent],
+    root: &Utf8PathBuf,
+    ignored_dirs: &[Utf8PathBuf],
+    gitignore: &Gitignore,
+) -> Vec<Utf8PathBuf> {
+    let mut out = BTreeSet::new();
+    for event in events {
+        for p in &event.paths {
+            if ignored_dirs
+                .iter()
+                .any(|dir| p.starts_with(dir.as_std_path()))
+            {
+                continue;
+            }
+            if gitignore.matched(p, p.is_dir()).is_ignore() {
+                continue;
+            }
+            let Ok(p) = Utf8PathBuf::from_path_buf(p.clone()) else {
+                continue;
+            };
+            if p.starts_with(root) {
+                out.insert(p);
+            }
+        }
+    }
+    out.into_iter().collect()
+}