feat(core): watcher + LSP improvements (#5283)

crates/biome_cli/src/commands/daemon.rs

@@ -5,7 +5,10 @@ use crate::{
 use biome_console::{ConsoleExt, markup};
 use biome_fs::OsFileSystem;
 use biome_lsp::ServerFactory;
-use biome_service::{TransportError, WorkspaceError, WorkspaceWatcher, workspace::WorkspaceClient};
+use biome_service::{
+    TransportError, WatcherInstruction, WorkspaceError, WorkspaceWatcher,
+    workspace::WorkspaceClient,
+};
 use camino::{Utf8Path, Utf8PathBuf};
 use std::{env, fs};
 use tokio::io;
@@ -73,7 +76,7 @@ pub(crate) fn run_server(
 ) -> Result<(), CliDiagnostic> {
     setup_tracing_subscriber(log_path.as_deref(), log_file_name_prefix.as_deref());
 
-    let (mut watcher, instruction_channel, _) = WorkspaceWatcher::new()?;
+    let (mut watcher, instruction_channel) = WorkspaceWatcher::new()?;
 
     let rt = Runtime::new()?;
     let factory = ServerFactory::new(stop_on_disconnect, instruction_channel.sender.clone());
@@ -99,6 +102,7 @@ pub(crate) fn run_server(
             }
             _ = cancellation.notified() => {
                 tracing::info!("Received shutdown signal");
+                let _ = instruction_channel.sender.send(WatcherInstruction::Stop);
                 Ok(())
             }
         }

crates/biome_cli/src/service/unix.rs

@@ -196,11 +196,11 @@ pub(crate) async fn print_socket() -> io::Result<()> {
 pub(crate) async fn run_daemon(factory: ServerFactory) -> io::Result<Infallible> {
     let path = get_socket_name();
 
-    info!("Trying to connect to socket {}", path.as_str());
+    info!("Trying to connect to socket {path}");
 
     // Try to remove the socket file if it already exists
     if path.exists() {
-        info!("Remove socket folder {}", path.as_str());
+        info!("Remove socket {path}");
         fs::remove_file(&path)?;
     }
 

crates/biome_dependency_graph/src/dependency_graph.rs

@@ -149,7 +149,7 @@ impl DependencyGraph {
                     .is_err()
                 {
                     break;
-                };
+                }
                 parent = path.parent();
             }
         }

crates/biome_lsp/src/handlers/text_document.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use crate::diagnostics::LspError;
 use crate::utils::apply_document_changes;
 use crate::{documents::Document, session::Session};
@@ -19,7 +21,7 @@ use tracing::{debug, error, field, info};
     )
 )]
 pub(crate) async fn did_open(
-    session: &Session,
+    session: &Arc<Session>,
     params: lsp_types::DidOpenTextDocumentParams,
 ) -> Result<(), LspError> {
     let url = params.text_document.uri;
@@ -41,7 +43,7 @@ pub(crate) async fn did_open(
                 path: parent_path.clone(),
                 open_uninitialized: true,
             })?;
-            session.insert_project(parent_path, project_key);
+            session.insert_and_scan_project(project_key, parent_path);
             project_key
         }
     };

crates/biome_lsp/src/server.rs

@@ -11,6 +11,7 @@ use biome_diagnostics::panic::PanicError;
 use biome_fs::{ConfigName, FileSystem, MemoryFileSystem, OsFileSystem};
 use biome_service::workspace::{
     CloseProjectParams, OpenProjectParams, RageEntry, RageParams, RageResult,
+    ServiceDataNotification,
 };
 use biome_service::{WatcherInstruction, WorkspaceServer};
 use crossbeam::channel::{Sender, bounded};
@@ -22,7 +23,7 @@ use std::panic::RefUnwindSafe;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use tokio::io::{AsyncRead, AsyncWrite};
-use tokio::sync::Notify;
+use tokio::sync::{Notify, watch};
 use tokio::task::spawn_blocking;
 use tower_lsp::jsonrpc::Result as LspResult;
 use tower_lsp::{ClientSocket, lsp_types::*};
@@ -398,11 +399,7 @@ impl LanguageServer for LSPServer {
                 match result {
                     Ok(project_key) => {
                         self.session
-                            .insert_project(project_path.clone(), project_key);
-
-                        self.session
-                            .scan_project_folder(project_key, project_path)
-                            .await;
+                            .insert_and_scan_project(project_key, project_path.clone());
 
                         self.session.update_all_diagnostics().await;
                     }
@@ -553,6 +550,12 @@ pub struct ServerFactory {
     /// This shared flag is set to true once at least one sessions has been
     /// initialized on this server instance
     is_initialized: Arc<AtomicBool>,
+
+    /// Receiver for service data notifications.
+    ///
+    /// If we receive a notification here, diagnostics for open documents are
+    /// all refreshed.
+    service_data_rx: watch::Receiver<ServiceDataNotification>,
 }
 
 impl Default for ServerFactory {
@@ -564,29 +567,34 @@ impl Default for ServerFactory {
 impl ServerFactory {
     /// Regular constructor for use in the daemon.
     pub fn new(stop_on_disconnect: bool, instruction_tx: Sender<WatcherInstruction>) -> Self {
+        let (service_data_tx, service_data_rx) = watch::channel(ServiceDataNotification::Updated);
         Self {
             cancellation: Arc::default(),
             workspace: Arc::new(WorkspaceServer::new(
                 Box::new(OsFileSystem::default()),
                 instruction_tx,
+                service_data_tx,
             )),
             sessions: Sessions::default(),
             next_session_key: AtomicU64::new(0),
             stop_on_disconnect,
             is_initialized: Arc::default(),
+            service_data_rx,
         }
     }
 
     /// Constructor for use in tests.
     pub fn new_with_fs(fs: Box<dyn FileSystem>) -> Self {
-        let (tx, _) = bounded(0);
+        let (watcher_tx, _) = bounded(0);
+        let (service_data_tx, service_data_rx) = watch::channel(ServiceDataNotification::Updated);
         Self {
             cancellation: Arc::default(),
-            workspace: Arc::new(WorkspaceServer::new(fs, tx)),
+            workspace: Arc::new(WorkspaceServer::new(fs, watcher_tx, service_data_tx)),
             sessions: Sessions::default(),
             next_session_key: AtomicU64::new(0),
             stop_on_disconnect: true,
             is_initialized: Arc::default(),
+            service_data_rx,
         }
     }
 
@@ -597,7 +605,13 @@ impl ServerFactory {
         let session_key = SessionKey(self.next_session_key.fetch_add(1, Ordering::Relaxed));
 
         let mut builder = LspService::build(move |client| {
-            let session = Session::new(session_key, client, workspace, self.cancellation.clone());
+            let session = Session::new(
+                session_key,
+                client,
+                workspace,
+                self.cancellation.clone(),
+                self.service_data_rx.clone(),
+            );
             let handle = Arc::new(session);
 
             let mut sessions = self.sessions.lock().unwrap();

crates/biome_lsp/src/server.tests.rs

@@ -52,6 +52,17 @@ macro_rules! url {
     };
 }
 
+macro_rules! clear_notifications {
+    ($channel:expr) => {
+        if $channel
+            .has_changed()
+            .expect("Channel should not be closed")
+        {
+            let _ = $channel.changed().await;
+        }
+    };
+}
+
 fn fixable_diagnostic(line: u32) -> Result<lsp::Diagnostic> {
     Ok(lsp::Diagnostic {
         range: Range {
@@ -2992,9 +3003,9 @@ export function bar() {
     fs.create_file("foo.ts", FOO_CONTENT);
     fs.create_file("bar.ts", BAR_CONTENT);
 
-    let (mut watcher, instruction_channel, notification_channel) = WorkspaceWatcher::new()?;
+    let (mut watcher, instruction_channel) = WorkspaceWatcher::new()?;
 
-    let factory = ServerFactory::new(true, instruction_channel.sender.clone());
+    let mut factory = ServerFactory::new(true, instruction_channel.sender.clone());
 
     let workspace = factory.workspace();
     tokio::task::spawn_blocking(move || {
@@ -3062,14 +3073,15 @@ export function bar() {
         "This import is part of a cycle."
     );
 
-    let _ = notification_channel.receiver.try_recv(); // Clear notification, if any.
+    clear_notifications!(factory.service_data_rx);
 
     // ARRANGE: Remove `bar.ts`.
     std::fs::remove_file(fs.working_directory.join("bar.ts")).expect("Cannot remove bar.ts");
 
-    notification_channel
-        .receiver
-        .recv()
+    factory
+        .service_data_rx
+        .changed()
+        .await
         .expect("Expected notification");
 
     // ACT: Pull diagnostics.
@@ -3094,13 +3106,14 @@ export function bar() {
     assert_eq!(result.diagnostics.len(), 0);
 
     // ARRANGE: Recreate `bar.ts`.
-    let _ = notification_channel.receiver.try_recv(); // Clear notification, if any.
+    clear_notifications!(factory.service_data_rx);
 
     fs.create_file("bar.ts", BAR_CONTENT);
 
-    notification_channel
-        .receiver
-        .recv()
+    factory
+        .service_data_rx
+        .changed()
+        .await
         .expect("Expected notification");
 
     // ACT: Pull diagnostics.
@@ -3129,13 +3142,14 @@ export function bar() {
     );
 
     // ARRANGE: Fix `bar.ts`.
-    let _ = notification_channel.receiver.try_recv(); // Clear notification, if any.
+    clear_notifications!(factory.service_data_rx);
 
     fs.create_file("bar.ts", BAR_CONTENT_FIXED);
 
-    notification_channel
-        .receiver
-        .recv()
+    factory
+        .service_data_rx
+        .changed()
+        .await
         .expect("Expected notification");
 
     // ACT: Pull diagnostics.
@@ -3159,7 +3173,6 @@ export function bar() {
     // ASSERT: Diagnostic should disappear again with a fixed `bar.ts`.
     assert_eq!(result.diagnostics.len(), 0);
 
-    let _ = instruction_channel.sender.send(WatcherInstruction::Stop);
     server.shutdown().await?;
     reader.abort();
 
@@ -3201,9 +3214,9 @@ export function bar() {
     fs.create_file("foo.ts", FOO_CONTENT);
     fs.create_file("utils/bar.ts", BAR_CONTENT);
 
-    let (mut watcher, instruction_channel, notification_channel) = WorkspaceWatcher::new()?;
+    let (mut watcher, instruction_channel) = WorkspaceWatcher::new()?;
 
-    let factory = ServerFactory::new(true, instruction_channel.sender.clone());
+    let mut factory = ServerFactory::new(true, instruction_channel.sender.clone());
 
     let workspace = factory.workspace();
     tokio::task::spawn_blocking(move || {
@@ -3271,7 +3284,7 @@ export function bar() {
         "This import is part of a cycle."
     );
 
-    let _ = notification_channel.receiver.try_recv(); // Clear notification, if any.
+    clear_notifications!(factory.service_data_rx);
 
     // ARRANGE: Move `utils` directory.
     std::fs::rename(
@@ -3280,9 +3293,10 @@ export function bar() {
     )
     .expect("Cannot move utils");
 
-    notification_channel
-        .receiver
-        .recv()
+    factory
+        .service_data_rx
+        .changed()
+        .await
         .expect("Expected notification");
 
     // ACT: Pull diagnostics.
@@ -3308,17 +3322,18 @@ export function bar() {
     assert_eq!(result.diagnostics.len(), 0);
 
     // ARRANGE: Move `utils` back.
-    let _ = notification_channel.receiver.try_recv(); // Clear notification, if any.
+    clear_notifications!(factory.service_data_rx);
 
     std::fs::rename(
         fs.working_directory.join("bin"),
         fs.working_directory.join("utils"),
     )
     .expect("Cannot restore utils");
 
-    notification_channel
-        .receiver
-        .recv()
+    factory
+        .service_data_rx
+        .changed()
+        .await
         .expect("Expected notification");
 
     // ACT: Pull diagnostics.
@@ -3346,7 +3361,6 @@ export function bar() {
         "This import is part of a cycle."
     );
 
-    let _ = instruction_channel.sender.send(WatcherInstruction::Stop);
     server.shutdown().await?;
     reader.abort();