From d2146736b9e17afde765b0db6c7742f4e3d6c01b Mon Sep 17 00:00:00 2001 From: gustav-fff <286169375+gustav-fff@users.noreply.github.com> Date: Fri, 16 Oct 2026 07:50:08 -0700 Subject: [PATCH 1/2] fix: batch FSEvents watches and stop blocking pool broadcasts (#938) notify's FsEventWatcher restarts its stream on every watch() and can lose a CFRunLoopStop issued right after a start, leaving the caller in pthread_join forever. create_debouncer hit that twice per watcher (base, .git, .git/logs). Register all paths with one update_paths call on recursive backends. The stuck pool worker then deadlocked every later scan through the blocking BACKGROUND_THREAD_POOL.broadcast in hint_allocator_collect and release_thread_buffers. Use spawn_broadcast so cleanup never waits on a busy worker. Closes #938 --- crates/fff-core/src/file_picker.rs | 3 +- crates/fff-core/src/index/bigram_filter.rs | 29 ++++++++++- .../src/watcher/background_watcher.rs | 52 ++++++++++++------- 3 files changed, 63 insertions(+), 21 deletions(-) diff --git a/crates/fff-core/src/file_picker.rs b/crates/fff-core/src/file_picker.rs index 847726507..312638acc 100644 --- a/crates/fff-core/src/file_picker.rs +++ b/crates/fff-core/src/file_picker.rs @@ -2470,7 +2470,8 @@ pub(crate) fn hint_allocator_collect() { // Collect BACKGROUND_THREAD_POOL workers — that's where the bigram // builder allocated memory. `rayon::broadcast` would target the global // pool, which is the wrong set of threads. - BACKGROUND_THREAD_POOL.broadcast(|_| unsafe { libmimalloc_sys::mi_collect(true) }); + // Non-blocking: a blocking broadcast stalls the scan behind any busy or stuck worker. + BACKGROUND_THREAD_POOL.spawn_broadcast(|_| unsafe { libmimalloc_sys::mi_collect(true) }); // Main thread too. unsafe { libmimalloc_sys::mi_collect(true) }; diff --git a/crates/fff-core/src/index/bigram_filter.rs b/crates/fff-core/src/index/bigram_filter.rs index c670851af..ca04bb480 100644 --- a/crates/fff-core/src/index/bigram_filter.rs +++ b/crates/fff-core/src/index/bigram_filter.rs @@ -1209,7 +1209,7 @@ pub(crate) fn release_thread_buffers() { READ_BUF.with_borrow_mut(|buf| *buf = Vec::new()); NORM_BUF.with_borrow_mut(|buf| *buf = Vec::new()); } - crate::parallelism::BACKGROUND_THREAD_POOL.broadcast(|_| release()); + crate::parallelism::BACKGROUND_THREAD_POOL.spawn_broadcast(|_| release()); release(); } @@ -1674,4 +1674,31 @@ mod tests { // Files without the needle stay filtered out. assert!(!BigramFilter::is_candidate(&candidates, 5)); } + + // A stuck pool worker (e.g. a hung FSEvents join) must not stall the caller (#938). + #[test] + fn release_thread_buffers_does_not_wait_for_busy_workers() { + use std::sync::mpsc; + use std::time::Duration; + + let (started_tx, started_rx) = mpsc::channel::<()>(); + let (unblock_tx, unblock_rx) = mpsc::channel::<()>(); + crate::parallelism::BACKGROUND_THREAD_POOL.spawn(move || { + started_tx.send(()).unwrap(); + let _ = unblock_rx.recv(); + }); + started_rx.recv().unwrap(); + + let (done_tx, done_rx) = mpsc::channel::<()>(); + std::thread::spawn(move || { + release_thread_buffers(); + let _ = done_tx.send(()); + }); + let finished = done_rx.recv_timeout(Duration::from_secs(5)).is_ok(); + unblock_tx.send(()).unwrap(); + assert!( + finished, + "release_thread_buffers blocked on a busy pool worker" + ); + } } diff --git a/crates/fff-core/src/watcher/background_watcher.rs b/crates/fff-core/src/watcher/background_watcher.rs index af4e7561f..e728ef88a 100644 --- a/crates/fff-core/src/watcher/background_watcher.rs +++ b/crates/fff-core/src/watcher/background_watcher.rs @@ -8,7 +8,7 @@ use crate::sort_buffer::sort_with_buffer; use crate::watch::{RawWatchEvent, WatchEventKind}; use git2::Repository; use notify::event::{AccessKind, AccessMode, ModifyKind, RenameMode}; -use notify::{Config, EventKind, EventKindMask, RecursiveMode}; +use notify::{Config, EventKind, EventKindMask, PathOp, RecursiveMode}; use notify_debouncer_full::{DebounceEventResult, DebouncedEvent, NoCache, new_debouncer_opt}; use parking_lot::Mutex; use std::path::{Path, PathBuf}; @@ -234,8 +234,24 @@ impl BackgroundWatcher { config, )?; + // The .git directory is excluded from the file list but we still need + // to observe changes that affect git status (staging, unstaging, + // committing, branch switches, merges, etc) + let git_paths = git_status_paths(git_workdir.as_ref()); + if use_recursive { - debouncer.watch(base_path.as_path(), RecursiveMode::Recursive)?; + // FSEvents restarts the stream on every watch() and can lose a stop issued right after + // a start, hanging the join forever (notify-rs/notify#942). Register paths in one batch. + let mut ops = vec![PathOp::watch_recursive(base_path.clone())]; + ops.extend(git_paths.into_iter().map(PathOp::watch_non_recursive)); + if let Err(e) = debouncer.update_paths(ops) { + match e.origin { + Some(ref op) if op.as_path() != base_path => { + warn!("Failed to watch {}: {}", op.as_path().display(), e.source) + } + _ => return Err(e.source.into()), + } + } info!( "File watcher initialized with single recursive watch on {}", base_path.display(), @@ -288,12 +304,13 @@ impl BackgroundWatcher { path = ?base_path.display(), "File watcher initialized" ); - } - // The .git directory is excluded from the file list but we still need - // to observe changes that affect git status (staging, unstaging, - // committing, branch switches, merges, etc) - watch_git_status_paths(&mut debouncer, git_workdir.as_ref()); + for path in git_paths { + if let Err(e) = debouncer.watch(&path, RecursiveMode::NonRecursive) { + warn!("Failed to watch {}: {}", path.display(), e); + } + } + } Ok(debouncer) } @@ -1022,33 +1039,30 @@ fn is_dotgit_change_affecting_status(changed: &Path, repo: &Option) false } -fn watch_git_status_paths(debouncer: &mut Debouncer, git_workdir: Option<&PathBuf>) { +fn git_status_paths(git_workdir: Option<&PathBuf>) -> Vec { let Some(workdir) = git_workdir else { - return; + return Vec::new(); }; let git_dir = workdir.join(".git"); if !git_dir.is_dir() { - return; + return Vec::new(); } // We have tried to be smart about the internal git state but // it appeared more harmful that it's worth it, so we just watch // for the most obvious paths like HEAD, MERGE_HEAD, index.lock - if let Err(e) = debouncer.watch(&git_dir, RecursiveMode::NonRecursive) { - warn!("Failed to watch .git directory: {}", e); - } - - // `.git` above is non-recursive, so on Linux (per-dir inotify watches) + // + // `.git` is non-recursive, so on Linux (per-dir inotify watches) // events for `logs/HEAD` — the commit-finished signal used by // `is_dotgit_change_affecting_status` — would never be delivered without // watching `.git/logs` itself. On macOS/Windows the recursive base watch // already covers it; an extra watch is harmless there. let logs_dir = git_dir.join("logs"); - if logs_dir.is_dir() - && let Err(e) = debouncer.watch(&logs_dir, RecursiveMode::NonRecursive) - { - warn!("Failed to watch .git/logs directory: {}", e); + if logs_dir.is_dir() { + vec![git_dir, logs_dir] + } else { + vec![git_dir] } } From 0b349fdd1ff539005dd786160e540a4ca98b6068 Mon Sep 17 00:00:00 2001 From: gustav-fff <286169375+gustav-fff@users.noreply.github.com> Date: Fri, 16 Oct 2026 20:34:42 -0700 Subject: [PATCH 2/2] refactor(watcher): drop Vec allocations for git status watch paths git_status_paths now returns an Option-chain iterator, the FSEvents batch chains it straight into update_paths, and the inotify branch goes back to watch_git_status_paths. --- .../src/watcher/background_watcher.rs | 45 ++++++++----------- 1 file changed, 19 insertions(+), 26 deletions(-) diff --git a/crates/fff-core/src/watcher/background_watcher.rs b/crates/fff-core/src/watcher/background_watcher.rs index e728ef88a..2eff83e5f 100644 --- a/crates/fff-core/src/watcher/background_watcher.rs +++ b/crates/fff-core/src/watcher/background_watcher.rs @@ -234,16 +234,11 @@ impl BackgroundWatcher { config, )?; - // The .git directory is excluded from the file list but we still need - // to observe changes that affect git status (staging, unstaging, - // committing, branch switches, merges, etc) - let git_paths = git_status_paths(git_workdir.as_ref()); - if use_recursive { // FSEvents restarts the stream on every watch() and can lose a stop issued right after // a start, hanging the join forever (notify-rs/notify#942). Register paths in one batch. - let mut ops = vec![PathOp::watch_recursive(base_path.clone())]; - ops.extend(git_paths.into_iter().map(PathOp::watch_non_recursive)); + let ops = std::iter::once(PathOp::watch_recursive(base_path.clone())) + .chain(git_status_paths(git_workdir.as_ref()).map(PathOp::watch_non_recursive)); if let Err(e) = debouncer.update_paths(ops) { match e.origin { Some(ref op) if op.as_path() != base_path => { @@ -305,11 +300,10 @@ impl BackgroundWatcher { "File watcher initialized" ); - for path in git_paths { - if let Err(e) = debouncer.watch(&path, RecursiveMode::NonRecursive) { - warn!("Failed to watch {}: {}", path.display(), e); - } - } + // The .git directory is excluded from the file list but we still need + // to observe changes that affect git status (staging, unstaging, + // committing, branch switches, merges, etc) + watch_git_status_paths(&mut debouncer, git_workdir.as_ref()); } Ok(debouncer) @@ -1039,16 +1033,15 @@ fn is_dotgit_change_affecting_status(changed: &Path, repo: &Option) false } -fn git_status_paths(git_workdir: Option<&PathBuf>) -> Vec { - let Some(workdir) = git_workdir else { - return Vec::new(); - }; - - let git_dir = workdir.join(".git"); - if !git_dir.is_dir() { - return Vec::new(); +fn watch_git_status_paths(debouncer: &mut Debouncer, git_workdir: Option<&PathBuf>) { + for path in git_status_paths(git_workdir) { + if let Err(e) = debouncer.watch(&path, RecursiveMode::NonRecursive) { + warn!("Failed to watch {}: {}", path.display(), e); + } } +} +fn git_status_paths(git_workdir: Option<&PathBuf>) -> impl Iterator { // We have tried to be smart about the internal git state but // it appeared more harmful that it's worth it, so we just watch // for the most obvious paths like HEAD, MERGE_HEAD, index.lock @@ -1058,12 +1051,12 @@ fn git_status_paths(git_workdir: Option<&PathBuf>) -> Vec { // `is_dotgit_change_affecting_status` — would never be delivered without // watching `.git/logs` itself. On macOS/Windows the recursive base watch // already covers it; an extra watch is harmless there. - let logs_dir = git_dir.join("logs"); - if logs_dir.is_dir() { - vec![git_dir, logs_dir] - } else { - vec![git_dir] - } + let git_dir = git_workdir.map(|w| w.join(".git")).filter(|d| d.is_dir()); + let logs_dir = git_dir + .as_ref() + .map(|d| d.join("logs")) + .filter(|d| d.is_dir()); + git_dir.into_iter().chain(logs_dir) } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]