refactor(lsp): publish diagnostics independently (#10525)

Resolves #10518

cli/bench/lsp.rs

@@ -266,6 +266,10 @@ fn bench_big_file_edits(deno_exe: &Path) -> Result<Duration, AnyError> {
     }),
   )?;
 
+  let (method, _): (String, Option<Value>) = client.read_notification()?;
+  assert_eq!(method, "textDocument/publishDiagnostics");
+  let (method, _): (String, Option<Value>) = client.read_notification()?;
+  assert_eq!(method, "textDocument/publishDiagnostics");
   let (method, _): (String, Option<Value>) = client.read_notification()?;
   assert_eq!(method, "textDocument/publishDiagnostics");
 
@@ -324,6 +328,10 @@ fn bench_startup_shutdown(deno_exe: &Path) -> Result<Duration, AnyError> {
     }),
   )?;
 
+  let (method, _): (String, Option<Value>) = client.read_notification()?;
+  assert_eq!(method, "textDocument/publishDiagnostics");
+  let (method, _): (String, Option<Value>) = client.read_notification()?;
+  assert_eq!(method, "textDocument/publishDiagnostics");
   let (method, _): (String, Option<Value>) = client.read_notification()?;
   assert_eq!(method, "textDocument/publishDiagnostics");
 

cli/lsp/diagnostics.rs

@@ -22,6 +22,7 @@ use std::collections::HashMap;
 use std::collections::HashSet;
 use std::mem;
 use std::sync::Arc;
+use std::sync::Mutex;
 use std::thread;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
@@ -51,10 +52,10 @@ enum DiagnosticRequest {
 /// to the client.
 async fn publish_diagnostics(
   client: &Client,
-  collection: &mut DiagnosticCollection,
+  collection: Arc<Mutex<DiagnosticCollection>>,
   snapshot: &language_server::StateSnapshot,
 ) {
-  let mark = snapshot.performance.mark("publish_diagnostics");
+  let mut collection = collection.lock().unwrap();
   let maybe_changes = collection.take_changes();
   if let Some(diagnostic_changes) = maybe_changes {
     for specifier in diagnostic_changes {
@@ -89,13 +90,11 @@ async fn publish_diagnostics(
       client.publish_diagnostics(uri, diagnostics, version).await;
     }
   }
-
-  snapshot.performance.measure(mark);
 }
 
 async fn update_diagnostics(
   client: &Client,
-  collection: &mut DiagnosticCollection,
+  collection: Arc<Mutex<DiagnosticCollection>>,
   snapshot: &language_server::StateSnapshot,
   ts_server: &tsc::TsServer,
 ) {
@@ -105,66 +104,46 @@ async fn update_diagnostics(
   };
 
   let mark = snapshot.performance.mark("update_diagnostics");
+
   let lint = async {
-    let mut diagnostics = None;
+    let collection = collection.clone();
     if lint_enabled {
-      let mark = snapshot.performance.mark("prepare_diagnostics_lint");
-      diagnostics = Some(
-        generate_lint_diagnostics(snapshot.clone(), collection.clone()).await,
+      let mark = snapshot.performance.mark("update_diagnostics_lint");
+      let diagnostics =
+        generate_lint_diagnostics(snapshot.clone(), collection.clone()).await;
+      {
+        let mut collection = collection.lock().unwrap();
+        for (specifier, version, diagnostics) in diagnostics {
+          collection.set(
+            specifier,
+            DiagnosticSource::Lint,
+            version,
+            diagnostics,
           );
+        }
+      }
+      publish_diagnostics(client, collection, snapshot).await;
       snapshot.performance.measure(mark);
     };
-    Ok::<_, AnyError>(diagnostics)
   };
 
   let ts = async {
-    let mut diagnostics = None;
     if enabled {
-      let mark = snapshot.performance.mark("prepare_diagnostics_ts");
-      diagnostics = Some(
-        generate_ts_diagnostics(
+      let collection = collection.clone();
+      let mark = snapshot.performance.mark("update_diagnostics_ts");
+      let diagnostics = generate_ts_diagnostics(
         snapshot.clone(),
         collection.clone(),
         ts_server,
       )
-        .await?,
-      );
-      snapshot.performance.measure(mark);
-    };
-    Ok::<_, AnyError>(diagnostics)
-  };
-
-  let deps = async {
-    let mut diagnostics = None;
-    if enabled {
-      let mark = snapshot.performance.mark("prepare_diagnostics_deps");
-      diagnostics = Some(
-        generate_dependency_diagnostics(snapshot.clone(), collection.clone())
-          .await?,
-      );
-      snapshot.performance.measure(mark);
-    };
-    Ok::<_, AnyError>(diagnostics)
-  };
-
-  let (lint_res, ts_res, deps_res) = tokio::join!(lint, ts, deps);
-  let mut disturbed = false;
-
-  match lint_res {
-    Ok(Some(diagnostics)) => {
-      for (specifier, version, diagnostics) in diagnostics {
-        collection.set(specifier, DiagnosticSource::Lint, version, diagnostics);
-        disturbed = true;
-      }
-    }
-    Err(err) => {
-      error!("Internal error: {}", err);
-    }
-    _ => (),
-  }
-
-  match ts_res {
-    Ok(Some(diagnostics)) => {
+      .await
+      .map_err(|err| {
+        error!("Error generating TypeScript diagnostics: {}", err);
+        err
+      })
+      .unwrap_or_default();
+      {
+        let mut collection = collection.lock().unwrap();
         for (specifier, version, diagnostics) in diagnostics {
           collection.set(
             specifier,
@@ -172,33 +151,45 @@ async fn update_diagnostics(
             version,
             diagnostics,
           );
-        disturbed = true;
         }
       }
-    Err(err) => {
-      error!("Internal error: {}", err);
-    }
-    _ => (),
-  }
-
-  match deps_res {
-    Ok(Some(diagnostics)) => {
-      for (specifier, version, diagnostics) in diagnostics {
-        collection.set(specifier, DiagnosticSource::Deno, version, diagnostics);
-        disturbed = true;
-      }
-    }
-    Err(err) => {
-      error!("Internal error: {}", err);
-    }
-    _ => (),
-  }
+      publish_diagnostics(client, collection, snapshot).await;
       snapshot.performance.measure(mark);
+    };
+  };
 
-  if disturbed {
-    publish_diagnostics(client, collection, snapshot).await
+  let deps = async {
+    if enabled {
+      let collection = collection.clone();
+      let mark = snapshot.performance.mark("update_diagnostics_deps");
+      let diagnostics =
+        generate_dependency_diagnostics(snapshot.clone(), collection.clone())
+          .await
+          .map_err(|err| {
+            error!("Error generating dependency diagnostics: {}", err);
+            err
+          })
+          .unwrap_or_default();
+      {
+        let mut collection = collection.lock().unwrap();
+        for (specifier, version, diagnostics) in diagnostics {
+          collection.set(
+            specifier,
+            DiagnosticSource::Deno,
+            version,
+            diagnostics,
+          );
         }
       }
+      publish_diagnostics(client, collection, snapshot).await;
+      snapshot.performance.measure(mark);
+    };
+  };
+
+  tokio::join!(lint, ts, deps);
+
+  snapshot.performance.measure(mark);
+}
 
 /// A server which calculates diagnostics in its own thread and publishes them
 /// to an LSP client.
@@ -223,7 +214,7 @@ impl DiagnosticsServer {
 
     let _join_handle = thread::spawn(move || {
       let runtime = create_basic_runtime();
-      let mut collection = DiagnosticCollection::default();
+      let collection = Arc::new(Mutex::new(DiagnosticCollection::default()));
 
       runtime.block_on(async {
         // Debounce timer delay. 150ms between keystrokes is about 45 WPM, so we
@@ -253,6 +244,8 @@ impl DiagnosticsServer {
                 None => break, // Request channel closed.
                 Some(Get(specifier, source, tx)) => {
                   let diagnostics = collection
+                    .lock()
+                    .unwrap()
                     .diagnostics_for(&specifier, &source)
                     .cloned()
                     .collect();
@@ -260,7 +253,7 @@ impl DiagnosticsServer {
                   let _ = tx.send(diagnostics);
                 }
                 Some(Invalidate(specifier)) => {
-                  collection.invalidate(&specifier);
+                  collection.lock().unwrap().invalidate(&specifier);
                 }
                 Some(Update) => {
                   dirty = true;
@@ -275,7 +268,7 @@ impl DiagnosticsServer {
               let snapshot = language_server.lock().await.snapshot();
               update_diagnostics(
                 &client,
-                &mut collection,
+                collection.clone(),
                 &snapshot,
                 &ts_server
               ).await;
@@ -321,7 +314,7 @@ impl DiagnosticsServer {
 #[derive(Debug, Default, Clone)]
 struct DiagnosticCollection {
   map: HashMap<(ModuleSpecifier, DiagnosticSource), Vec<lsp::Diagnostic>>,
-  versions: HashMap<ModuleSpecifier, i32>,
+  versions: HashMap<ModuleSpecifier, HashMap<DiagnosticSource, i32>>,
   changes: HashSet<ModuleSpecifier>,
 }
 
@@ -333,9 +326,12 @@ impl DiagnosticCollection {
     version: Option<i32>,
     diagnostics: Vec<lsp::Diagnostic>,
   ) {
-    self.map.insert((specifier.clone(), source), diagnostics);
+    self
+      .map
+      .insert((specifier.clone(), source.clone()), diagnostics);
     if let Some(version) = version {
-      self.versions.insert(specifier.clone(), version);
+      let source_versions = self.versions.entry(specifier.clone()).or_default();
+      source_versions.insert(source, version);
     }
     self.changes.insert(specifier);
   }
@@ -352,8 +348,16 @@ impl DiagnosticCollection {
       .flatten()
   }
 
-  pub fn get_version(&self, specifier: &ModuleSpecifier) -> Option<i32> {
-    self.versions.get(specifier).cloned()
+  pub fn get_version(
+    &self,
+    specifier: &ModuleSpecifier,
+    source: &DiagnosticSource,
+  ) -> Option<i32> {
+    if let Some(source_versions) = self.versions.get(specifier) {
+      source_versions.get(source).cloned()
+    } else {
+      None
+    }
   }
 
   pub fn invalidate(&mut self, specifier: &ModuleSpecifier) {
@@ -373,14 +377,16 @@ pub type DiagnosticVec =
 
 async fn generate_lint_diagnostics(
   state_snapshot: language_server::StateSnapshot,
-  collection: DiagnosticCollection,
+  collection: Arc<Mutex<DiagnosticCollection>>,
 ) -> DiagnosticVec {
   tokio::task::spawn_blocking(move || {
     let mut diagnostic_list = Vec::new();
-
     for specifier in state_snapshot.documents.open_specifiers() {
       let version = state_snapshot.documents.version(specifier);
-      let current_version = collection.get_version(specifier);
+      let current_version = collection
+        .lock()
+        .unwrap()
+        .get_version(specifier, &DiagnosticSource::Lint);
       if version != current_version {
         let media_type = MediaType::from(specifier);
         if let Ok(Some(source_code)) =
@@ -520,18 +526,22 @@ fn ts_json_to_diagnostics(
 
 async fn generate_ts_diagnostics(
   state_snapshot: language_server::StateSnapshot,
-  collection: DiagnosticCollection,
+  collection: Arc<Mutex<DiagnosticCollection>>,
   ts_server: &tsc::TsServer,
 ) -> Result<DiagnosticVec, AnyError> {
   let mut diagnostics = Vec::new();
   let mut specifiers = Vec::new();
+  {
+    let collection = collection.lock().unwrap();
     for specifier in state_snapshot.documents.open_specifiers() {
       let version = state_snapshot.documents.version(specifier);
-    let current_version = collection.get_version(specifier);
+      let current_version =
+        collection.get_version(specifier, &DiagnosticSource::TypeScript);
       if version != current_version {
         specifiers.push(specifier.clone());
       }
     }
+  }
   if !specifiers.is_empty() {
     let req = tsc::RequestMethod::GetDiagnostics(specifiers);
     let res = ts_server.request(state_snapshot.clone(), req).await?;
@@ -551,7 +561,7 @@ async fn generate_ts_diagnostics(
 
 async fn generate_dependency_diagnostics(
   mut state_snapshot: language_server::StateSnapshot,
-  collection: DiagnosticCollection,
+  collection: Arc<Mutex<DiagnosticCollection>>,
 ) -> Result<DiagnosticVec, AnyError> {
   tokio::task::spawn_blocking(move || {
     let mut diagnostics = Vec::new();
@@ -559,7 +569,7 @@ async fn generate_dependency_diagnostics(
     let sources = &mut state_snapshot.sources;
     for specifier in state_snapshot.documents.open_specifiers() {
       let version = state_snapshot.documents.version(specifier);
-      let current_version = collection.get_version(specifier);
+      let current_version = collection.lock().unwrap().get_version(specifier, &DiagnosticSource::Deno);
       if version != current_version {
         let mut diagnostic_list = Vec::new();
         if let Some(dependencies) = state_snapshot.documents.dependencies(specifier) {