fix(vfs): keep macOS watcher setup from blocking readiness

crates/vfs/src/notify.rs

@@ -4,17 +4,32 @@ use std::{
     fs,
     path::{Component, Path},
     sync::atomic::AtomicUsize,
+    time::Duration,
 };
 
 use crossbeam_channel::{Receiver, Sender, select, unbounded};
-use notify::{Config, EventKind, RecommendedWatcher, RecursiveMode, Watcher, event::AccessKind};
+use notify::{Config, EventKind, RecursiveMode, Watcher, event::AccessKind};
 use rayon::iter::{IndexedParallelIterator as _, IntoParallelIterator as _, ParallelIterator};
 use rustc_hash::FxHashSet;
 use utils::paths::{AbsPath, AbsPathBuf, Utf8PathBuf};
 use walkdir::WalkDir;
 
 use crate::loader::{self, LoadingProgress};
 
+// FSEvents watcher registration can block while starting or stopping its
+// CFRunLoop, which must never hold workspace readiness hostage. PollWatcher has
+// bounded setup and shutdown behavior and preserves recursive change detection
+// on macOS.
+#[cfg(target_os = "macos")]
+type BackendWatcher = notify::PollWatcher;
+#[cfg(not(target_os = "macos"))]
+type BackendWatcher = notify::RecommendedWatcher;
+
+#[cfg(target_os = "macos")]
+const WATCHER_BACKEND: &str = "poll";
+#[cfg(not(target_os = "macos"))]
+const WATCHER_BACKEND: &str = "recommended";
+
 #[derive(Debug)]
 pub struct NotifyHandle {
     // Relative order of fields below is significant.
@@ -60,7 +75,7 @@ struct NotifyActor {
     watched_dir_entries: Vec<loader::Directories>,
     seen_paths: FxHashSet<AbsPathBuf>,
     // Drop order is significant.
-    watcher: Option<(RecommendedWatcher, Receiver<NotifyEvent>)>,
+    watcher: Option<(BackendWatcher, Receiver<NotifyEvent>)>,
 }
 
 #[derive(Debug)]
@@ -100,15 +115,25 @@ impl NotifyActor {
                         self.watcher = None;
                         if !config.watch.is_empty() {
                             let (watcher_sender, watcher_receiver) = unbounded();
-                            let watcher = log_notify_error(RecommendedWatcher::new(
+                            let watcher_config =
+                                Config::default().with_poll_interval(Duration::from_secs(1));
+                            let watcher = BackendWatcher::new(
                                 move |event| {
-                                    // we don't care about the error. If sending fails that usually
-                                    // means we were dropped, so unwrapping will just add to the
-                                    // panic noise.
+                                    // A disconnected receiver means the actor was dropped. Do not
+                                    // panic in the platform callback because that only obscures the
+                                    // shutdown cause.
                                     _ = watcher_sender.send(event);
                                 },
-                                Config::default(),
-                            ));
+                                watcher_config,
+                            )
+                            .map_err(|error| {
+                                tracing::error!(
+                                    %error,
+                                    backend = WATCHER_BACKEND,
+                                    "failed to create file watcher"
+                                );
+                            })
+                            .ok();
                             self.watcher = watcher.map(|it| (it, watcher_receiver));
                         }
 
@@ -192,11 +217,17 @@ impl NotifyActor {
                     }
                 },
                 Event::NotifyEvent(event) => {
-                    if let Some(event) = log_notify_error(event)
-                        && let EventKind::Create(_)
-                        | EventKind::Modify(_)
-                        | EventKind::Remove(_)
-                        | EventKind::Access(AccessKind::Open(_)) = event.kind
+                    let event = match event {
+                        Ok(event) => event,
+                        Err(error) => {
+                            tracing::error!(%error, backend = WATCHER_BACKEND, "file watcher error");
+                            continue;
+                        }
+                    };
+                    if let EventKind::Create(_)
+                    | EventKind::Modify(_)
+                    | EventKind::Remove(_)
+                    | EventKind::Access(AccessKind::Open(_)) = event.kind
                     {
                         let abs_paths: Vec<AbsPathBuf> = event
                             .paths
@@ -352,8 +383,15 @@ impl NotifyActor {
     }
 
     fn watch(&mut self, path: &Path) {
-        if let Some((watcher, _)) = &mut self.watcher {
-            log_notify_error(watcher.watch(path, RecursiveMode::Recursive));
+        if let Some((watcher, _)) = &mut self.watcher
+            && let Err(error) = watcher.watch(path, RecursiveMode::Recursive)
+        {
+            tracing::error!(
+                %error,
+                path = %path.display(),
+                backend = WATCHER_BACKEND,
+                "failed to register file watcher path"
+            );
         }
     }
 
@@ -367,10 +405,6 @@ fn read(path: &AbsPath) -> Option<Vec<u8>> {
     std::fs::read(path).ok()
 }
 
-fn log_notify_error<T>(res: notify::Result<T>) -> Option<T> {
-    res.map_err(|err| tracing::warn!("notify error: {}", err)).ok()
-}
-
 /// Is `path` a symlink to a parent directory?
 ///
 /// Including this path is guaranteed to cause an infinite loop. This
@@ -388,3 +422,67 @@ fn path_might_be_cyclic(path: &Path) -> bool {
 
     is_relative_parent || path.starts_with(destination)
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::{Duration, Instant};
+
+    use crossbeam_channel::unbounded;
+    use tempfile::tempdir;
+    use utils::paths::{AbsPathBuf, Utf8PathBuf};
+
+    use super::NotifyHandle;
+    use crate::loader::{self, Handle as _, LoadingProgress};
+
+    #[test]
+    fn watcher_setup_finishes_loading() {
+        let temp_dir = tempdir().unwrap();
+        std::fs::write(temp_dir.path().join("top.sv"), "module top; endmodule\n").unwrap();
+        let root = AbsPathBuf::try_from(
+            Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).unwrap(),
+        )
+        .unwrap();
+        let directories = loader::Directories {
+            extensions: vec!["sv".to_owned()],
+            include: vec![root],
+            exclude: Vec::new(),
+        };
+
+        let config = loader::Config {
+            version: 7,
+            load: vec![loader::Entry::Directories(directories)],
+            watch: vec![0],
+        };
+
+        let (sender, receiver) = unbounded();
+        let mut handle = NotifyHandle::spawn(sender);
+        handle.set_config(config);
+
+        let deadline = Instant::now() + Duration::from_secs(5);
+        let mut loaded_batches = 0;
+        let mut journal = Vec::new();
+        loop {
+            let message = receiver.recv_deadline(deadline).unwrap_or_else(|error| {
+                panic!(
+                    "VFS configuration did not finish before the deadline: {error}; events: {journal:#?}"
+                )
+            });
+            journal.push(format!("{message:?}"));
+            match message {
+                loader::Message::Loaded { .. } => loaded_batches += 1,
+                loader::Message::Progress {
+                    config_version: 7,
+                    n_done: LoadingProgress::Finished,
+                    ..
+                } => break,
+                _ => {}
+            }
+        }
+
+        assert_eq!(loaded_batches, 1);
+
+        // The previous macOS FSEvents watcher could wait indefinitely for its run loop
+        // when this handle was dropped.
+        drop(handle);
+    }
+}

src/config.rs

@@ -84,6 +84,7 @@ pub struct Config {
     pub(crate) user_config: UserConfig,
     diagnostics_config: DiagnosticsConfig,
     pub(crate) project_manifests: Vec<ProjectManifest>,
+    main_loop_threads_num: usize,
 }
 
 #[derive(Debug, Clone)]
@@ -110,9 +111,17 @@ impl Config {
             user_config,
             diagnostics_config,
             project_manifests,
+            main_loop_threads_num: num_cpus::get_physical(),
         }
     }
 
+    #[cfg(test)]
+    pub(crate) fn with_main_loop_threads_num(mut self, threads: usize) -> Self {
+        assert!(threads > 0, "the main loop worker count must be greater than zero");
+        self.main_loop_threads_num = threads;
+        self
+    }
+
     pub(crate) fn update(&mut self, json: serde_json::Value) -> Result<(), ConfigError> {
         let (user_config, _snippets, errors) = Self::parse_initialization_options(json);
         let diagnostics_config = self.updated_diagnostics_config(&user_config);
@@ -167,7 +176,7 @@ impl Config {
     }
 
     pub fn main_loop_threads_num(&self) -> usize {
-        num_cpus::get_physical()
+        self.main_loop_threads_num
     }
 
     pub fn files(&self) -> FilesConfig {

src/global_state/main_loop.rs

@@ -99,7 +99,8 @@ mod tests {
             I18n::default(),
             UserConfig::default(),
             Vec::new(),
-        );
+        )
+        .with_main_loop_threads_num(1);
 
         let (server, client) = Connection::memory();
         (GlobalState::new(server.sender, config, lsp_types::TraceValue::Off), client)

src/tests.rs

@@ -31,8 +31,8 @@ use lsp_types::{
         Completion as CompletionRequest, DocumentDiagnosticRequest, DocumentSymbolRequest,
         ExecuteCommand, FoldingRangeRequest, Formatting, GotoDefinition, GotoTypeDefinition,
         HoverRequest, InlayHintRequest, References, Request as _, SemanticTokensFullRequest,
-        Shutdown, SignatureHelpRequest, WorkspaceConfiguration, WorkspaceDiagnosticRequest,
-        WorkspaceSymbolRequest,
+        Shutdown, SignatureHelpRequest, UnregisterCapability, WorkspaceConfiguration,
+        WorkspaceDiagnosticRequest, WorkspaceSymbolRequest,
     },
 };
 use serde::de::DeserializeOwned;
@@ -92,6 +92,7 @@ fn handle_test_server_request(client: &Connection, request: Request, context: &s
     if request.method == lsp_types::request::WorkDoneProgressCreate::METHOD
         || request.method == lsp_types::request::WorkspaceDiagnosticRefresh::METHOD
         || request.method == lsp_types::request::RegisterCapability::METHOD
+        || request.method == UnregisterCapability::METHOD
     {
         client
             .sender
@@ -121,11 +122,48 @@ fn test_server_config(
     client_caps: ClientCapabilities,
     user_config: UserConfig,
 ) -> config::Config {
-    test_server_config_with_i18n(root_path, client_caps, user_config, I18n::default())
+    test_server_config_with_roots_and_i18n(
+        root_path.clone(),
+        vec![root_path],
+        client_caps,
+        user_config,
+        I18n::default(),
+    )
 }
 
 fn test_server_config_with_i18n(
     root_path: AbsPathBuf,
+    client_caps: ClientCapabilities,
+    user_config: UserConfig,
+    i18n: I18n,
+) -> config::Config {
+    test_server_config_with_roots_and_i18n(
+        root_path.clone(),
+        vec![root_path],
+        client_caps,
+        user_config,
+        i18n,
+    )
+}
+
+fn test_server_config_with_roots(
+    root_path: AbsPathBuf,
+    workspace_roots: Vec<AbsPathBuf>,
+    client_caps: ClientCapabilities,
+    user_config: UserConfig,
+) -> config::Config {
+    test_server_config_with_roots_and_i18n(
+        root_path,
+        workspace_roots,
+        client_caps,
+        user_config,
+        I18n::default(),
+    )
+}
+
+fn test_server_config_with_roots_and_i18n(
+    root_path: AbsPathBuf,
+    workspace_roots: Vec<AbsPathBuf>,
     mut client_caps: ClientCapabilities,
     user_config: UserConfig,
     i18n: I18n,
@@ -137,15 +175,8 @@ fn test_server_config_with_i18n(
         log_filename: None,
         profile_trace: None,
     };
-    config::Config::new(
-        opt,
-        root_path.clone(),
-        client_caps,
-        vec![root_path],
-        i18n,
-        user_config,
-        Vec::new(),
-    )
+    config::Config::new(opt, root_path, client_caps, workspace_roots, i18n, user_config, Vec::new())
+        .with_main_loop_threads_num(1)
 }
 
 fn spawn_test_workspace(