wsb-daemon 0.248.6265

//! Which directories a filesystem watcher may watch, and a watcher that
//! stays inside that scope.
//!
//! ## Why this exists
//!
//! On Linux, notify's inotify backend implements a recursive watch by adding
//! ONE KERNEL WATCH PER DIRECTORY, against a per-user budget
//! (`fs.inotify.max_user_watches`, commonly 65536). A recursive watch on a
//! repo therefore spends watches on every `node_modules/`, `target/` and
//! `build/` directory in it — hundreds of thousands of directories in this
//! workspace — even though the event CALLBACKS filter those paths out. The
//! budget ran out, repos failed to watch at all, and the very directories
//! nobody cares about were what spent it.
//!
//! Filtering must happen at the WATCH layer, not the event layer. The scope
//! here is git's own answer to "what matters": a change inside a git-ignored
//! directory cannot change `git status`, and an artifact tree a repo declares
//! ignorable is not something an editor needs conflict events for. So the
//! walk respects `.gitignore` / `.git/info/exclude`, never descends into
//! `.git` or `.wsb`, and (optionally) stops at nested-repository boundaries —
//! the fleet watcher watches each repo separately, so the superproject's
//! watcher crossing into submodules would watch every tree twice.
//!
//! Directories created after startup are picked up by the rescan loop: their
//! creation raises an event in the already-watched parent, which both nudges
//! the consumer and triggers a re-walk that adds the new directories to the
//! watch set.
//!
//! ## Adding directories in one batch
//!
//! Directories are added through `paths_mut`, never one `watch` at a time. On
//! macOS a watcher is ONE FSEvents stream over every path it holds, and a
//! single `watch` stops that stream and creates it again over all of them, so
//! adding n directories one by one costs n stream creations over up to n
//! paths. The daemon's start registers its workspace before it serves; with
//! the peer-continuity watcher's 1176 directories that took 65 seconds, the
//! command that started the daemon gave up waiting after 20, and the next one
//! started a second daemon (20 Oct 2026). A batch creates the stream once.
//!
//! ## One watcher, many roots
//!
//! A watcher takes a LIST of roots, not one, because the other inotify budget
//! is `fs.inotify.max_user_instances` — commonly 128 PER USER, shared with
//! every editor and tool the person is running. One watcher per repository
//! spent forty of those on this workspace alone, and a second wsb alongside an
//! editor then failed to watch anything at all ("Too many open files"). The
//! directories still cost one kernel watch each against the far larger
//! `max_user_watches`; the instances do not have to multiply with them.

use wsb::prelude::*;
use notify_debouncer_mini::{new_debouncer, notify::RecursiveMode, DebouncedEvent};
use std::collections::HashSet;
use std::sync::mpsc;
use std::sync::Mutex;
use std::time::Duration;

/// The directories a watcher over `root` should watch, each non-recursively.
pub struct WatchScope {
    pub directories: Vec<PathBuf>,
    /// Walk entries that could not be read (permissions, races). Counted so a
    /// degraded scope is reported, never silently smaller.
    pub walk_errors: usize,
}

/// Enumerate the watchable directories under `root`.
///
/// Skips: anything git-ignored (per-directory `.gitignore`, the repo's
/// `.git/info/exclude`, the global excludes file), `.git` and `.wsb` always,
/// and — when `cross_repo_boundaries` is false — any nested repository
/// (a subdirectory carrying its own `.git`), because that repository is
/// watched by its own watcher.
pub fn watchable_directories(root: &Path, cross_repo_boundaries: bool) -> WatchScope {
    let mut builder = ignore::WalkBuilder::new(root);
    builder
        .hidden(false) // .github and friends matter to git status
        .follow_links(false)
        .git_ignore(true)
        .git_global(true)
        .git_exclude(true);
    builder.filter_entry(move |entry| {
        let name = entry.file_name();
        if name == ".git" || name == ".wsb" {
            return false;
        }
        if !cross_repo_boundaries
            && entry.depth() > 0
            && entry.file_type().is_some_and(|t| t.is_dir())
            && entry.path().join(".git").exists()
        {
            return false;
        }
        true
    });

    let mut directories = Vec::new();
    let mut walk_errors = 0usize;
    for entry in builder.build() {
        match entry {
            Ok(entry) => {
                if entry.file_type().is_some_and(|t| t.is_dir()) {
                    directories.push(entry.into_path());
                }
            }
            Err(_) => walk_errors += 1,
        }
    }
    WatchScope {
        directories,
        walk_errors,
    }
}

/// The roots a watcher covers, for a message a person reads.
fn describe_roots(roots: &[PathBuf]) -> String {
    match roots {
        [] => "no roots".to_string(),
        [one] => one.display().to_string(),
        [first, rest @ ..] => format!("{} and {} more", first.display(), rest.len()),
    }
}

/// The scope of several roots at once, deduplicated.
///
/// Roots may nest — the superproject contains its submodules — and with
/// `cross_repo_boundaries` false their scopes are disjoint by construction.
/// The set is deduplicated anyway, because watching one directory twice in one
/// instance is a wasted kernel watch and a doubled event.
pub fn watchable_directories_of(roots: &[PathBuf], cross_repo_boundaries: bool) -> WatchScope {
    let mut seen = HashSet::new();
    let mut directories = Vec::new();
    let mut walk_errors = 0usize;
    for root in roots {
        let scope = watchable_directories(root, cross_repo_boundaries);
        walk_errors += scope.walk_errors;
        for dir in scope.directories {
            if seen.insert(dir.clone()) {
                directories.push(dir);
            }
        }
    }
    WatchScope {
        directories,
        walk_errors,
    }
}

/// Spawn a process-lifetime watcher over `roots`, watching exactly the scope
/// above and re-walking whenever a new directory appears inside it.
///
/// ONE watcher — one inotify instance — however many roots are passed. Callers
/// that need to know which root an event belongs to match the event path
/// against their own roots; see the fleet watcher.
///
/// `on_events` runs on the notify thread with every debounced batch; it must
/// not block. Returns Err when the roots yield nothing watchable or the
/// watcher cannot be created — a watcher that silently watched nothing would
/// report a healthy workspace forever.
pub fn spawn_scoped_watcher<F>(
    label: String,
    roots: Vec<PathBuf>,
    cross_repo_boundaries: bool,
    debounce: Duration,
    on_events: F,
) -> Result<(), String>
where
    F: Fn(&[DebouncedEvent]) + Send + 'static,
{
    let scope = watchable_directories_of(&roots, cross_repo_boundaries);
    if scope.directories.is_empty() {
        return Err(format!(
            "{label}: no watchable directories under {}",
            describe_roots(&roots)
        ));
    }

    // The set the callback consults to spot brand-new directories. Shared
    // with the rescan thread, which is the only writer after startup.
    let watched: Arc<Mutex<HashSet<PathBuf>>> = Arc::new(Mutex::new(HashSet::new()));
    let (rescan_tx, rescan_rx) = mpsc::channel::<()>();

    let callback_watched = Arc::clone(&watched);
    let callback_label = label.clone();
    let mut debouncer = new_debouncer(
        debounce,
        move |result: Result<Vec<DebouncedEvent>, notify_debouncer_mini::notify::Error>| {
            match result {
                Ok(events) => {
                    // A directory we do not watch yet means the scope grew —
                    // ask the owning thread to re-walk. The consumer still
                    // hears about the creation itself via on_events, because
                    // the PARENT directory is watched. An edited ignore file
                    // also re-walks: un-ignoring an existing directory must
                    // bring it into the scope, not leave it dark until the
                    // next restart.
                    let needs_rescan = {
                        let watched = callback_watched
                            .lock()
                            .expect("fs-scope watch set lock poisoned");
                        events.iter().any(|event| {
                            (event.path.is_dir() && !watched.contains(&event.path))
                                || event
                                    .path
                                    .file_name()
                                    .is_some_and(|name| name == ".gitignore" || name == ".ignore")
                        })
                    };
                    if needs_rescan {
                        let _ = rescan_tx.send(());
                    }
                    on_events(&events);
                }
                Err(error) => {
                    tracing::warn!(watcher = %callback_label, error = %error, "notify error");
                }
            }
        },
    )
    .map_err(|e| format!("{label}: cannot create watcher: {e}"))?;

    // Watch the initial scope on the caller's thread, so a total failure is
    // an Err the caller sees rather than a log line nobody does.
    let mut initial_failures = 0usize;
    let mut first_error: Option<String> = None;
    {
        let mut watched = watched.lock().expect("fs-scope watch set lock poisoned");
        let mut batch = debouncer.watcher().paths_mut();
        for dir in &scope.directories {
            match batch.add(dir, RecursiveMode::NonRecursive) {
                Ok(()) => {
                    watched.insert(dir.clone());
                }
                Err(error) => {
                    initial_failures += 1;
                    if first_error.is_none() {
                        first_error = Some(format!("{error}"));
                    }
                }
            }
        }
        batch
            .commit()
            .map_err(|e| format!("{label}: cannot start watching {}: {e}", describe_roots(&roots)))?;
        if watched.is_empty() {
            return Err(format!(
                "{label}: could not watch any of {} directories under {} — first error: {}",
                scope.directories.len(),
                describe_roots(&roots),
                first_error.unwrap_or_else(|| "none recorded".to_string()),
            ));
        }
        tracing::info!(
            watcher = %label,
            roots = roots.len(),
            watched = watched.len(),
            failed = initial_failures,
            walk_errors = scope.walk_errors,
            "fs scope watching"
        );
        if initial_failures > 0 {
            tracing::warn!(
                watcher = %label,
                failed = initial_failures,
                error = first_error.as_deref().unwrap_or("none recorded"),
                "fs scope: some directories could not be watched; the periodic \
                 scan still covers them, at its own cadence"
            );
        }
    }

    // The rescan thread owns the debouncer for the rest of the process: it
    // adds watches for directories that appear later, and keeping the
    // debouncer alive here is what keeps the notify thread running.
    let thread_label = label.clone();
    std::thread::Builder::new()
        .name(format!("fs-scope:{label}"))
        .spawn(move || {
            while rescan_rx.recv().is_ok() {
                // Coalesce a burst (an unpacked tarball is thousands of
                // directory-creations) into one walk.
                while rescan_rx.try_recv().is_ok() {}
                let scope = watchable_directories_of(&roots, cross_repo_boundaries);
                let mut added = 0usize;
                let mut failures = 0usize;
                {
                    let mut watched =
                        watched.lock().expect("fs-scope watch set lock poisoned");
                    let new: Vec<PathBuf> = scope
                        .directories
                        .into_iter()
                        .filter(|dir| !watched.contains(dir))
                        .collect();
                    // Only a scope that grew opens a batch: on macOS opening
                    // one stops the stream, and a re-walk that found nothing
                    // new must not restart it.
                    if !new.is_empty() {
                        let mut batch = debouncer.watcher().paths_mut();
                        let mut accepted = Vec::new();
                        for dir in new {
                            match batch.add(&dir, RecursiveMode::NonRecursive) {
                                Ok(()) => accepted.push(dir),
                                Err(_) => failures += 1,
                            }
                        }
                        match batch.commit() {
                            Ok(()) => {
                                added = accepted.len();
                                watched.extend(accepted);
                            }
                            Err(error) => {
                                failures += accepted.len();
                                tracing::warn!(
                                    watcher = %thread_label,
                                    %error,
                                    "fs scope: the new directories could not be watched"
                                );
                            }
                        }
                    }
                }
                if added > 0 || failures > 0 {
                    tracing::info!(
                        watcher = %thread_label,
                        added,
                        failures,
                        "fs scope grew"
                    );
                }
            }
            // Every sender is gone — the callback (and so the debouncer) is
            // only dropped with this thread, so this is process shutdown.
        })
        .map_err(|e| format!("{label}: cannot spawn rescan thread: {e}"))?;

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A fake repo: `.git` marker, a tracked-ish tree, an ignored artifact
    /// tree, and a nested repository.
    fn scaffold() -> tempfile::TempDir {
        let dir = tempfile::tempdir().expect("tempdir");
        let root = dir.path();
        std::fs::create_dir(root.join(".git")).unwrap();
        std::fs::write(root.join(".gitignore"), "node_modules/\ntarget/\n").unwrap();
        std::fs::create_dir_all(root.join("src/deep")).unwrap();
        std::fs::create_dir_all(root.join("node_modules/left-pad")).unwrap();
        std::fs::create_dir_all(root.join("target/release")).unwrap();
        std::fs::create_dir_all(root.join(".wsb/state")).unwrap();
        std::fs::create_dir_all(root.join("sub/inner")).unwrap();
        std::fs::create_dir(root.join("sub/.git")).unwrap();
        dir
    }

    fn names(root: &Path, scope: &WatchScope) -> Vec<String> {
        let mut out: Vec<String> = scope
            .directories
            .iter()
            // With `/`, as the expectations are written; `display()` spells a
            // Windows path `src\deep` and matched none of them.
            .map(|d| wsb::platform::slash_path(d.strip_prefix(root).expect("scope stays under the root")))
            .collect();
        out.sort();
        out
    }

    /// TEST8968: git-ignored directories are outside the watch scope.
    ///
    /// This is the whole point of the module: a recursive watch spent one
    /// kernel watch per node_modules/target directory and exhausted
    /// fs.inotify.max_user_watches. If an ignored directory slips back into
    /// the scope, the budget is spent on trees whose changes cannot change
    /// git status.
    #[test]
    fn test8968_ignored_directories_are_not_watchable() {
        let dir = scaffold();
        let scope = watchable_directories(dir.path(), false);
        let listed = names(dir.path(), &scope);
        assert!(
            !listed.iter().any(|n| n.starts_with("node_modules")),
            "node_modules must be outside the scope: {listed:?}"
        );
        assert!(
            !listed.iter().any(|n| n.starts_with("target")),
            "target must be outside the scope: {listed:?}"
        );
        assert!(
            listed.contains(&"src".to_string()) && listed.contains(&"src/deep".to_string()),
            "the tracked tree must be inside the scope: {listed:?}"
        );
        assert!(
            listed.contains(&String::new()),
            "the root itself must be watched: {listed:?}"
        );
    }

    /// TEST8969: `.git` and `.wsb` are never watched — one is git's own
    /// churn, the other is wsb's, and both feed back into the thing the
    /// watcher exists to notice.
    #[test]
    fn test8969_git_and_wsb_are_never_watchable() {
        let dir = scaffold();
        for cross in [false, true] {
            let scope = watchable_directories(dir.path(), cross);
            let listed = names(dir.path(), &scope);
            assert!(
                !listed.iter().any(|n| n.starts_with(".git") || n.starts_with(".wsb")),
                "cross={cross}: .git/.wsb must be outside the scope: {listed:?}"
            );
        }
    }

    /// TEST8970: a nested repository is a boundary for a per-repo watcher
    /// (its own watcher covers it; crossing would watch every submodule
    /// twice), and part of the scope for a whole-workspace watcher.
    #[test]
    fn test8970_nested_repositories_are_a_boundary_unless_crossed() {
        let dir = scaffold();
        let bounded = names(dir.path(), &watchable_directories(dir.path(), false));
        assert!(
            !bounded.iter().any(|n| n.starts_with("sub")),
            "a nested repo must be outside a bounded scope: {bounded:?}"
        );
        let crossed = names(dir.path(), &watchable_directories(dir.path(), true));
        assert!(
            crossed.contains(&"sub".to_string()) && crossed.contains(&"sub/inner".to_string()),
            "a crossing scope must include the nested repo: {crossed:?}"
        );
    }

    /// TEST8971: an empty scope is a refusal, not a silent no-op watcher.
    #[test]
    fn test8971_watching_nothing_is_an_error() {
        let dir = tempfile::tempdir().expect("tempdir");
        let missing = dir.path().join("does-not-exist");
        let result = spawn_scoped_watcher(
            "test".to_string(),
            vec![missing],
            false,
            Duration::from_millis(10),
            |_| {},
        );
        assert!(
            result.is_err(),
            "a root with nothing watchable must be refused"
        );
    }

    /// TEST9710: many roots cost ONE inotify instance, not one each.
    ///
    /// This is the budget that ran out in practice: watches are capped at
    /// ~65536 per user but INSTANCES at 128, shared with every editor and tool
    /// the person runs. A watcher per repository spent forty of them on this
    /// workspace, and the next wsb to start could not watch anything at all —
    /// "Too many open files" — falling back to the periodic scan without
    /// anyone asking it to.
    #[test]
    fn test9710_many_roots_are_one_watcher() {
        let first = scaffold();
        let second = scaffold();
        let roots = vec![first.path().to_path_buf(), second.path().to_path_buf()];

        let scope = watchable_directories_of(&roots, false);
        // Both trees are in the one scope...
        assert!(
            scope.directories.contains(&first.path().join("src")),
            "the first root's tree is missing: {:?}",
            scope.directories
        );
        assert!(
            scope.directories.contains(&second.path().join("src/deep")),
            "the second root's tree is missing: {:?}",
            scope.directories
        );
        // ...and each directory appears once, since watching one twice in one
        // instance spends a kernel watch to deliver a doubled event.
        let unique: HashSet<&PathBuf> = scope.directories.iter().collect();
        assert_eq!(
            unique.len(),
            scope.directories.len(),
            "a directory was listed twice"
        );

        // A single instance carries both, which is the whole point.
        let seen = std::sync::Arc::new(Mutex::new(Vec::<PathBuf>::new()));
        let recorder = std::sync::Arc::clone(&seen);
        spawn_scoped_watcher(
            "test-multi".to_string(),
            roots,
            false,
            Duration::from_millis(10),
            move |events| {
                let mut seen = recorder.lock().expect("recorder");
                seen.extend(events.iter().map(|e| e.path.clone()));
            },
        )
        .expect("one watcher must cover every root");

        std::fs::write(first.path().join("src/one.txt"), "a").unwrap();
        std::fs::write(second.path().join("src/two.txt"), "b").unwrap();

        // Compared against the CANONICAL roots. On macOS the temporary
        // directory is reached through a symlink — `/var` is `/private/var` —
        // so `TempDir::path()` is `/var/folders/…` while the watcher reports
        // the resolved `/private/var/folders/…`. `starts_with` is textual, so
        // it never matched and this failed every run on a mac while reporting
        // the two paths it was looking at as though they were absent.
        let first_root = wsb::platform::canonicalize(&first.path()).expect("the first root resolves");
        let second_root = wsb::platform::canonicalize(&second.path()).expect("the second root resolves");

        // The debounce is 10ms; give the notify thread room without racing on
        // an exact deadline.
        for _ in 0..50 {
            std::thread::sleep(Duration::from_millis(40));
            let seen = seen.lock().expect("recorder");
            let from_first = seen.iter().any(|p| p.starts_with(&first_root));
            let from_second = seen.iter().any(|p| p.starts_with(&second_root));
            if from_first && from_second {
                return;
            }
        }
        let seen = seen.lock().expect("recorder");
        panic!("one watcher did not report both roots: {seen:?}");
    }

    /// TEST12915: a scope of thousands of directories is watched in seconds,
    /// and its last directory is watched as surely as its first.
    ///
    /// Added one `watch` at a time, a macOS watcher recreated its FSEvents
    /// stream over every path it held for each directory, and a daemon whose
    /// workspace has 1176 watchable directories took 65 seconds to start —
    /// longer than the command that started it would wait. In one batch the
    /// stream is created once. Two thousand directories one by one take tens of
    /// seconds on a Mac; the bound is far above what a batch takes and far
    /// below that.
    #[test]
    fn test12915_a_large_scope_is_watched_in_one_batch() {
        let dir = tempfile::tempdir().expect("tempdir");
        let root = dir.path();
        std::fs::create_dir(root.join(".git")).unwrap();
        const DIRECTORIES: usize = 2000;
        for i in 0..DIRECTORIES {
            std::fs::create_dir(root.join(format!("d{i:04}"))).unwrap();
        }

        let seen = std::sync::Arc::new(Mutex::new(Vec::<PathBuf>::new()));
        let recorder = std::sync::Arc::clone(&seen);
        let started = std::time::Instant::now();
        spawn_scoped_watcher(
            "test-large".to_string(),
            vec![root.to_path_buf()],
            false,
            Duration::from_millis(10),
            move |events| {
                let mut seen = recorder.lock().expect("recorder");
                seen.extend(events.iter().map(|e| e.path.clone()));
            },
        )
        .expect("a large scope must be watched");
        let took = started.elapsed();
        assert!(
            took < Duration::from_secs(10),
            "watching {DIRECTORIES} directories took {took:?}"
        );

        let last = wsb::platform::canonicalize(root.join(format!("d{:04}", DIRECTORIES - 1)))
            .expect("the last directory resolves");
        std::fs::write(last.join("touched.txt"), "x").unwrap();
        for _ in 0..50 {
            std::thread::sleep(Duration::from_millis(40));
            if seen.lock().expect("recorder").iter().any(|p| p.starts_with(&last)) {
                return;
            }
        }
        let seen = seen.lock().expect("recorder");
        panic!("a change in the last directory added was not reported: {seen:?}");
    }
}