fix(lsp): diagnostics use own thread and debounce (#9572)

4 files changed, 395 insertions, 198 deletions

diff --git a/cli/lsp/config.rs b/cli/lsp/config.rs
index cb814d0fd..8d31e3d54 100644
--- a/cli/lsp/config.rs
+++ b/cli/lsp/config.rs
@@ -15,7 +15,7 @@ pub struct ClientCapabilities {
   pub workspace_did_change_watched_files: bool,
 }
 
-#[derive(Debug, Default, Deserialize)]
+#[derive(Debug, Default, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct CodeLensSettings {
   /// Flag for providing implementation code lenses.
@@ -30,7 +30,7 @@ pub struct CodeLensSettings {
   pub references_all_functions: bool,
 }
 
-#[derive(Debug, Default, Deserialize)]
+#[derive(Debug, Default, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct WorkspaceSettings {
   pub enable: bool,
@@ -81,7 +81,7 @@ impl WorkspaceSettings {
   }
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct Config {
   pub client_capabilities: ClientCapabilities,
   pub root_uri: Option<Url>,
diff --git a/cli/lsp/diagnostics.rs b/cli/lsp/diagnostics.rs
index 705f0866d..1bfb19867 100644
--- a/cli/lsp/diagnostics.rs
+++ b/cli/lsp/diagnostics.rs
@@ -3,20 +3,33 @@
 use super::analysis::get_lint_references;
 use super::analysis::references_to_diagnostics;
 use super::analysis::ResolvedDependency;
-use super::language_server::StateSnapshot;
+use super::language_server;
 use super::tsc;
 
 use crate::diagnostics;
 use crate::media_type::MediaType;
+use crate::tokio_util::create_basic_runtime;
 
+use deno_core::error::anyhow;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::ModuleSpecifier;
 use lspower::lsp;
+use lspower::Client;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::mem;
+use std::sync::Arc;
+use std::thread;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio::time;
+
+// 150ms between keystrokes is about 45 WPM, so we want something that is longer
+// than that, but not too long to introduce detectable UI delay.  200ms is a
+// decent compromise.
+const DIAGNOSTIC_DEBOUNCE_MS: u64 = 200;
 
 #[derive(Debug, Clone, Hash, PartialEq, Eq)]
 pub enum DiagnosticSource {
@@ -25,8 +38,311 @@ pub enum DiagnosticSource {
   TypeScript,
 }
 
+#[derive(Debug)]
+enum DiagnosticRequest {
+  Get(
+    ModuleSpecifier,
+    DiagnosticSource,
+    oneshot::Sender<Vec<lsp::Diagnostic>>,
+  ),
+  Invalidate(ModuleSpecifier),
+  Update,
+}
+
+/// Given a client and a diagnostics collection, publish the appropriate changes
+/// to the client.
+async fn publish_diagnostics(
+  client: &Client,
+  collection: &mut DiagnosticCollection,
+  snapshot: &language_server::StateSnapshot,
+) {
+  let mark = snapshot.performance.mark("publish_diagnostics");
+  let maybe_changes = collection.take_changes();
+  if let Some(diagnostic_changes) = maybe_changes {
+    for specifier in diagnostic_changes {
+      // TODO(@kitsonk) not totally happy with the way we collect and store
+      // different types of diagnostics and offer them up to the client, we
+      // do need to send "empty" vectors though when a particular feature is
+      // disabled, otherwise the client will not clear down previous
+      // diagnostics
+      let mut diagnostics: Vec<lsp::Diagnostic> =
+        if snapshot.config.settings.lint {
+          collection
+            .diagnostics_for(&specifier, &DiagnosticSource::Lint)
+            .cloned()
+            .collect()
+        } else {
+          vec![]
+        };
+      if snapshot.config.settings.enable {
+        diagnostics.extend(
+          collection
+            .diagnostics_for(&specifier, &DiagnosticSource::TypeScript)
+            .cloned(),
+        );
+        diagnostics.extend(
+          collection
+            .diagnostics_for(&specifier, &DiagnosticSource::Deno)
+            .cloned(),
+        );
+      }
+      let uri = specifier.clone();
+      let version = snapshot.documents.version(&specifier);
+      client.publish_diagnostics(uri, diagnostics, version).await;
+    }
+  }
+
+  snapshot.performance.measure(mark);
+}
+
+async fn update_diagnostics(
+  client: &Client,
+  collection: &mut DiagnosticCollection,
+  snapshot: &language_server::StateSnapshot,
+  ts_server: &tsc::TsServer,
+) {
+  let (enabled, lint_enabled) = {
+    let config = &snapshot.config;
+    (config.settings.enable, config.settings.lint)
+  };
+
+  let mark = snapshot.performance.mark("update_diagnostics");
+  let lint = async {
+    let mut diagnostics = None;
+    if lint_enabled {
+      let mark = snapshot.performance.mark("prepare_diagnostics_lint");
+      diagnostics = Some(
+        generate_lint_diagnostics(snapshot.clone(), collection.clone()).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(
+          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)) => {
+      for (specifier, version, diagnostics) in diagnostics {
+        collection.set(
+          specifier,
+          DiagnosticSource::TypeScript,
+          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);
+    }
+    _ => (),
+  }
+  snapshot.performance.measure(mark);
+
+  if disturbed {
+    publish_diagnostics(client, collection, snapshot).await
+  }
+}
+
+fn handle_request(
+  maybe_request: Option<DiagnosticRequest>,
+  collection: &mut DiagnosticCollection,
+  dirty: &mut bool,
+) -> bool {
+  match maybe_request {
+    Some(request) => {
+      match request {
+        DiagnosticRequest::Get(specifier, source, tx) => {
+          let diagnostics = collection
+            .diagnostics_for(&specifier, &source)
+            .cloned()
+            .collect();
+          if tx.send(diagnostics).is_err() {
+            error!("DiagnosticServer unable to send response on channel.");
+          }
+        }
+        DiagnosticRequest::Invalidate(specifier) => {
+          collection.invalidate(&specifier)
+        }
+        DiagnosticRequest::Update => *dirty = true,
+      }
+      true
+    }
+    _ => false,
+  }
+}
+
+/// A server which calculates diagnostics in its own thread and publishes them
+/// to an LSP client.
+#[derive(Debug)]
+pub(crate) struct DiagnosticsServer(
+  Option<mpsc::UnboundedSender<DiagnosticRequest>>,
+);
+
+impl DiagnosticsServer {
+  pub(crate) fn new() -> Self {
+    Self(None)
+  }
+
+  pub(crate) fn start(
+    &mut self,
+    language_server: Arc<tokio::sync::Mutex<language_server::Inner>>,
+    client: Client,
+    ts_server: Arc<tsc::TsServer>,
+  ) {
+    let (tx, mut rx) = mpsc::unbounded_channel::<DiagnosticRequest>();
+    self.0 = Some(tx);
+
+    let _join_handle = thread::spawn(move || {
+      let runtime = create_basic_runtime();
+      let mut collection = DiagnosticCollection::default();
+
+      runtime.block_on(async {
+        // Some(snapshot) is the flag we use to determine if something has
+        // changed where we will wait for the timeout of changes or a request
+        // that forces us to update diagnostics
+        let mut dirty = false;
+
+        loop {
+          let next = rx.recv();
+          tokio::pin!(next);
+
+          let duration = if dirty {
+            time::Duration::from_millis(DIAGNOSTIC_DEBOUNCE_MS)
+          } else {
+            // we need to await an arbitrary silly amount of time, so this is
+            // 1 year in seconds
+            time::Duration::new(31_622_400, 0)
+          };
+
+          // "race" the next message off the rx queue or the debounce timer.
+          // if the next message comes off the queue, the next iteration of the
+          // loop will reset the debounce future.  When the debounce future
+          // occurs, the diagnostics will be updated based on the snapshot that
+          // is retrieved, thereby "skipping" all the interim state changes.
+          tokio::select! {
+            _ = time::sleep(duration) => {
+              if dirty {
+                dirty = false;
+                let snapshot = {
+                  // make sure the lock drops
+                  language_server.lock().await.snapshot()
+                };
+                update_diagnostics(
+                  &client,
+                  &mut collection,
+                  &snapshot,
+                  &ts_server
+                ).await;
+              }
+              let maybe_request = next.await;
+              if !handle_request(maybe_request, &mut collection, &mut dirty) {
+                break;
+              }
+            }
+            maybe_request = &mut next => {
+              if !handle_request(maybe_request, &mut collection, &mut dirty) {
+                break;
+              }
+            }
+          }
+        }
+      })
+    });
+  }
+
+  pub async fn get(
+    &self,
+    specifier: ModuleSpecifier,
+    source: DiagnosticSource,
+  ) -> Result<Vec<lsp::Diagnostic>, AnyError> {
+    let (tx, rx) = oneshot::channel::<Vec<lsp::Diagnostic>>();
+    if let Some(self_tx) = &self.0 {
+      self_tx.send(DiagnosticRequest::Get(specifier, source, tx))?;
+      rx.await.map_err(|err| err.into())
+    } else {
+      Err(anyhow!("diagnostic server not started"))
+    }
+  }
+
+  pub fn invalidate(&self, specifier: ModuleSpecifier) -> Result<(), AnyError> {
+    if let Some(tx) = &self.0 {
+      tx.send(DiagnosticRequest::Invalidate(specifier))
+        .map_err(|err| err.into())
+    } else {
+      Err(anyhow!("diagnostic server not started"))
+    }
+  }
+
+  pub fn update(&self) -> Result<(), AnyError> {
+    if let Some(tx) = &self.0 {
+      tx.send(DiagnosticRequest::Update).map_err(|err| err.into())
+    } else {
+      Err(anyhow!("diagnostic server not started"))
+    }
+  }
+}
+
 #[derive(Debug, Default, Clone)]
-pub struct DiagnosticCollection {
+struct DiagnosticCollection {
   map: HashMap<(ModuleSpecifier, DiagnosticSource), Vec<lsp::Diagnostic>>,
   versions: HashMap<ModuleSpecifier, i32>,
   changes: HashSet<ModuleSpecifier>,
@@ -78,16 +394,16 @@ impl DiagnosticCollection {
 pub type DiagnosticVec =
   Vec<(ModuleSpecifier, Option<i32>, Vec<lsp::Diagnostic>)>;
 
-pub async fn generate_lint_diagnostics(
-  state_snapshot: StateSnapshot,
-  diagnostic_collection: DiagnosticCollection,
+async fn generate_lint_diagnostics(
+  state_snapshot: language_server::StateSnapshot,
+  collection: 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 = diagnostic_collection.get_version(specifier);
+      let current_version = collection.get_version(specifier);
       if version != current_version {
         let media_type = MediaType::from(specifier);
         if let Ok(Some(source_code)) =
@@ -229,16 +545,16 @@ fn ts_json_to_diagnostics(
     .collect()
 }
 
-pub async fn generate_ts_diagnostics(
-  state_snapshot: StateSnapshot,
-  diagnostic_collection: DiagnosticCollection,
+async fn generate_ts_diagnostics(
+  state_snapshot: language_server::StateSnapshot,
+  collection: DiagnosticCollection,
   ts_server: &tsc::TsServer,
 ) -> Result<DiagnosticVec, AnyError> {
   let mut diagnostics = Vec::new();
   let mut specifiers = Vec::new();
   for specifier in state_snapshot.documents.open_specifiers() {
     let version = state_snapshot.documents.version(specifier);
-    let current_version = diagnostic_collection.get_version(specifier);
+    let current_version = collection.get_version(specifier);
     if version != current_version {
       specifiers.push(specifier.clone());
     }
@@ -260,9 +576,9 @@ pub async fn generate_ts_diagnostics(
   Ok(diagnostics)
 }
 
-pub async fn generate_dependency_diagnostics(
-  mut state_snapshot: StateSnapshot,
-  diagnostic_collection: DiagnosticCollection,
+async fn generate_dependency_diagnostics(
+  mut state_snapshot: language_server::StateSnapshot,
+  collection: DiagnosticCollection,
 ) -> Result<DiagnosticVec, AnyError> {
   tokio::task::spawn_blocking(move || {
     let mut diagnostics = Vec::new();
@@ -270,7 +586,7 @@ pub 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 = diagnostic_collection.get_version(specifier);
+      let current_version = collection.get_version(specifier);
       if version != current_version {
         let mut diagnostic_list = Vec::new();
         if let Some(dependencies) = state_snapshot.documents.dependencies(specifier) {
diff --git a/cli/lsp/language_server.rs b/cli/lsp/language_server.rs
index 755151a24..96983dc52 100644
--- a/cli/lsp/language_server.rs
+++ b/cli/lsp/language_server.rs
@@ -41,7 +41,6 @@ use super::analysis::ResolvedDependency;
 use super::capabilities;
 use super::config::Config;
 use super::diagnostics;
-use super::diagnostics::DiagnosticCollection;
 use super::diagnostics::DiagnosticSource;
 use super::documents::DocumentCache;
 use super::performance::Performance;
@@ -66,6 +65,7 @@ pub struct LanguageServer(Arc<tokio::sync::Mutex<Inner>>);
 #[derive(Debug, Clone, Default)]
 pub struct StateSnapshot {
   pub assets: Assets,
+  pub config: Config,
   pub documents: DocumentCache,
   pub performance: Performance,
   pub sources: Sources,
@@ -80,8 +80,7 @@ pub(crate) struct Inner {
   client: Client,
   /// Configuration information.
   config: Config,
-  /// A collection of diagnostics from different sources.
-  diagnostics: DiagnosticCollection,
+  diagnostics_server: diagnostics::DiagnosticsServer,
   /// The "in-memory" documents in the editor which can be updated and changed.
   documents: DocumentCache,
   /// An optional URL which provides the location of a TypeScript configuration
@@ -100,7 +99,7 @@ pub(crate) struct Inner {
   /// A memoized version of fixable diagnostic codes retrieved from TypeScript.
   ts_fixable_diagnostics: Vec<String>,
   /// An abstraction that handles interactions with TypeScript.
-  ts_server: TsServer,
+  ts_server: Arc<TsServer>,
   /// A map of specifiers and URLs used to translate over the LSP.
   pub url_map: urls::LspUrlMap,
 }
@@ -118,21 +117,24 @@ impl Inner {
       .expect("could not access DENO_DIR");
     let location = dir.root.join("deps");
     let sources = Sources::new(&location);
+    let ts_server = Arc::new(TsServer::new());
+    let performance = Performance::default();
+    let diagnostics_server = diagnostics::DiagnosticsServer::new();
 
     Self {
       assets: Default::default(),
       client,
       config: Default::default(),
-      diagnostics: Default::default(),
+      diagnostics_server,
       documents: Default::default(),
       maybe_config_uri: Default::default(),
       maybe_import_map: Default::default(),
       maybe_import_map_uri: Default::default(),
       navigation_trees: Default::default(),
-      performance: Default::default(),
+      performance,
       sources,
       ts_fixable_diagnostics: Default::default(),
-      ts_server: TsServer::new(),
+      ts_server,
       url_map: Default::default(),
     }
   }
@@ -242,157 +244,10 @@ impl Inner {
     }
   }
 
-  async fn prepare_diagnostics(&mut self) -> Result<(), AnyError> {
-    let (enabled, lint_enabled) = {
-      let config = &self.config;
-      (config.settings.enable, config.settings.lint)
-    };
-
-    let lint = async {
-      let mut diagnostics = None;
-      if lint_enabled {
-        let mark = self.performance.mark("prepare_diagnostics_lint");
-        diagnostics = Some(
-          diagnostics::generate_lint_diagnostics(
-            self.snapshot(),
-            self.diagnostics.clone(),
-          )
-          .await,
-        );
-        self.performance.measure(mark);
-      };
-      Ok::<_, AnyError>(diagnostics)
-    };
-
-    let ts = async {
-      let mut diagnostics = None;
-      if enabled {
-        let mark = self.performance.mark("prepare_diagnostics_ts");
-        diagnostics = Some(
-          diagnostics::generate_ts_diagnostics(
-            self.snapshot(),
-            self.diagnostics.clone(),
-            &self.ts_server,
-          )
-          .await?,
-        );
-        self.performance.measure(mark);
-      };
-      Ok::<_, AnyError>(diagnostics)
-    };
-
-    let deps = async {
-      let mut diagnostics = None;
-      if enabled {
-        let mark = self.performance.mark("prepare_diagnostics_deps");
-        diagnostics = Some(
-          diagnostics::generate_dependency_diagnostics(
-            self.snapshot(),
-            self.diagnostics.clone(),
-          )
-          .await?,
-        );
-        self.performance.measure(mark);
-      };
-      Ok::<_, AnyError>(diagnostics)
-    };
-
-    let (lint_res, ts_res, deps_res) = tokio::join!(lint, ts, deps);
-    let mut disturbed = false;
-
-    if let Some(diagnostics) = lint_res? {
-      for (specifier, version, diagnostics) in diagnostics {
-        self.diagnostics.set(
-          specifier,
-          DiagnosticSource::Lint,
-          version,
-          diagnostics,
-        );
-        disturbed = true;
-      }
-    }
-
-    if let Some(diagnostics) = ts_res? {
-      for (specifier, version, diagnostics) in diagnostics {
-        self.diagnostics.set(
-          specifier,
-          DiagnosticSource::TypeScript,
-          version,
-          diagnostics,
-        );
-        disturbed = true;
-      }
-    }
-
-    if let Some(diagnostics) = deps_res? {
-      for (specifier, version, diagnostics) in diagnostics {
-        self.diagnostics.set(
-          specifier,
-          DiagnosticSource::Deno,
-          version,
-          diagnostics,
-        );
-        disturbed = true;
-      }
-    }
-
-    if disturbed {
-      self.publish_diagnostics().await?;
-    }
-
-    Ok(())
-  }
-
-  async fn publish_diagnostics(&mut self) -> Result<(), AnyError> {
-    let mark = self.performance.mark("publish_diagnostics");
-    let (maybe_changes, diagnostics_collection) = {
-      let diagnostics_collection = &mut self.diagnostics;
-      let maybe_changes = diagnostics_collection.take_changes();
-      (maybe_changes, diagnostics_collection.clone())
-    };
-    if let Some(diagnostic_changes) = maybe_changes {
-      for specifier in diagnostic_changes {
-        // TODO(@kitsonk) not totally happy with the way we collect and store
-        // different types of diagnostics and offer them up to the client, we
-        // do need to send "empty" vectors though when a particular feature is
-        // disabled, otherwise the client will not clear down previous
-        // diagnostics
-        let mut diagnostics: Vec<Diagnostic> = if self.config.settings.lint {
-          diagnostics_collection
-            .diagnostics_for(&specifier, &DiagnosticSource::Lint)
-            .cloned()
-            .collect()
-        } else {
-          vec![]
-        };
-        if self.enabled() {
-          diagnostics.extend(
-            diagnostics_collection
-              .diagnostics_for(&specifier, &DiagnosticSource::TypeScript)
-              .cloned(),
-          );
-          diagnostics.extend(
-            diagnostics_collection
-              .diagnostics_for(&specifier, &DiagnosticSource::Deno)
-              .cloned(),
-          );
-        }
-        let uri = specifier.clone();
-        let version = self.documents.version(&specifier);
-        self
-          .client
-          .publish_diagnostics(uri, diagnostics, version)
-          .await;
-      }
-    }
-
-    self.performance.measure(mark);
-    Ok(())
-  }
-
-  fn snapshot(&self) -> StateSnapshot {
+  pub(crate) fn snapshot(&self) -> StateSnapshot {
     StateSnapshot {
       assets: self.assets.clone(),
+      config: self.config.clone(),
       documents: self.documents.clone(),
       performance: self.performance.clone(),
       sources: self.sources.clone(),
@@ -667,8 +522,7 @@ impl Inner {
     self.analyze_dependencies(&specifier, &params.text_document.text);
     self.performance.measure(mark);
 
-    // TODO(@kitsonk): how to better lazily do this?
-    if let Err(err) = self.prepare_diagnostics().await {
+    if let Err(err) = self.diagnostics_server.update() {
       error!("{}", err);
     }
   }
@@ -687,8 +541,7 @@ impl Inner {
     }
     self.performance.measure(mark);
 
-    // TODO(@kitsonk): how to better lazily do this?
-    if let Err(err) = self.prepare_diagnostics().await {
+    if let Err(err) = self.diagnostics_server.update() {
       error!("{}", err);
     }
   }
@@ -706,8 +559,7 @@ impl Inner {
     self.navigation_trees.remove(&specifier);
 
     self.performance.measure(mark);
-    // TODO(@kitsonk): how to better lazily do this?
-    if let Err(err) = self.prepare_diagnostics().await {
+    if let Err(err) = self.diagnostics_server.update() {
       error!("{}", err);
     }
   }
@@ -755,7 +607,7 @@ impl Inner {
           .show_message(MessageType::Warning, err.to_string())
           .await;
       }
-      if let Err(err) = self.prepare_diagnostics().await {
+      if let Err(err) = self.diagnostics_server.update() {
         error!("{}", err);
       }
     } else {
@@ -931,11 +783,14 @@ impl Inner {
     }
     let line_index = self.get_line_index_sync(&specifier).unwrap();
     let mut code_actions = CodeActionCollection::default();
-    let file_diagnostics: Vec<Diagnostic> = self
-      .diagnostics
-      .diagnostics_for(&specifier, &DiagnosticSource::TypeScript)
-      .cloned()
-      .collect();
+    let file_diagnostics = self
+      .diagnostics_server
+      .get(specifier.clone(), DiagnosticSource::TypeScript)
+      .await
+      .map_err(|err| {
+        error!("Unable to get diagnostics: {}", err);
+        LspError::internal_error()
+      })?;
     for diagnostic in &fixable_diagnostics {
       match diagnostic.source.as_deref() {
         Some("deno-ts") => {
@@ -1748,7 +1603,13 @@ impl lspower::LanguageServer for LanguageServer {
     &self,
     params: InitializeParams,
   ) -> LspResult<InitializeResult> {
-    self.0.lock().await.initialize(params).await
+    let mut language_server = self.0.lock().await;
+    let client = language_server.client.clone();
+    let ts_server = language_server.ts_server.clone();
+    language_server
+      .diagnostics_server
+      .start(self.0.clone(), client, ts_server);
+    language_server.initialize(params).await
   }
 
   async fn initialized(&self, params: InitializedParams) {
@@ -1932,10 +1793,16 @@ impl Inner {
       if let Some(source) = self.documents.content(&referrer).unwrap() {
         self.analyze_dependencies(&referrer, &source);
       }
-      self.diagnostics.invalidate(&referrer);
+      self
+        .diagnostics_server
+        .invalidate(referrer)
+        .map_err(|err| {
+          error!("{}", err);
+          LspError::internal_error()
+        })?;
     }
 
-    self.prepare_diagnostics().await.map_err(|err| {
+    self.diagnostics_server.update().map_err(|err| {
       error!("{}", err);
       LspError::internal_error()
     })?;
@@ -2018,6 +1885,7 @@ mod tests {
     V: FnOnce(Value),
   {
     None,
+    Delay(u64),
     RequestAny,
     Request(u64, Value),
     RequestAssert(V),
@@ -2043,14 +1911,23 @@ mod tests {
         assert_eq!(self.service.poll_ready(), Poll::Ready(Ok(())));
         let fixtures_path = test_util::root_path().join("cli/tests/lsp");
         assert!(fixtures_path.is_dir());
-        let req_path = fixtures_path.join(req_path_str);
-        let req_str = fs::read_to_string(req_path).unwrap();
-        let req: jsonrpc::Incoming = serde_json::from_str(&req_str).unwrap();
         let response: Result<Option<jsonrpc::Outgoing>, ExitedError> =
-          self.service.call(req).await;
+          if req_path_str.is_empty() {
+            Ok(None)
+          } else {
+            let req_path = fixtures_path.join(req_path_str);
+            let req_str = fs::read_to_string(req_path).unwrap();
+            let req: jsonrpc::Incoming =
+              serde_json::from_str(&req_str).unwrap();
+            self.service.call(req).await
+          };
         match response {
           Ok(result) => match expected {
             LspResponse::None => assert_eq!(result, None),
+            LspResponse::Delay(millis) => {
+              tokio::time::sleep(tokio::time::Duration::from_millis(*millis))
+                .await
+            }
             LspResponse::RequestAny => match result {
               Some(jsonrpc::Outgoing::Response(_)) => (),
               _ => panic!("unexpected result: {:?}", result),
@@ -2296,18 +2173,18 @@ mod tests {
             "contents": [
               {
                 "language": "typescript",
-                "value": "const b: \"😃\"",
+                "value": "const b: \"🦕😃\"",
               },
               "",
             ],
             "range": {
               "start": {
                 "line": 2,
-                "character": 13,
+                "character": 15,
               },
               "end": {
                 "line": 2,
-                "character": 14,
+                "character": 16,
               },
             }
           }),
@@ -2418,7 +2295,7 @@ mod tests {
     let time = Instant::now();
     harness.run().await;
     assert!(
-      time.elapsed().as_millis() <= 15000,
+      time.elapsed().as_millis() <= 10000,
       "the execution time exceeded 10000ms"
     );
   }
@@ -2820,6 +2697,7 @@ mod tests {
       ("initialize_request.json", LspResponse::RequestAny),
       ("initialized_notification.json", LspResponse::None),
       ("did_open_notification_code_action.json", LspResponse::None),
+      ("", LspResponse::Delay(500)),
       (
         "code_action_request.json",
         LspResponse::RequestFixture(2, "code_action_response.json".to_string()),
@@ -2907,7 +2785,10 @@ mod tests {
         LspResponse::RequestAssert(|value| {
           let resp: PerformanceResponse =
             serde_json::from_value(value).unwrap();
-          assert_eq!(resp.result.averages.len(), 12);
+          // the len can be variable since some of the parts of the language
+          // server run in separate threads and may not add to performance by
+          // the time the results are checked.
+          assert!(resp.result.averages.len() >= 6);
         }),
       ),
       (
diff --git a/cli/tests/lsp/hover_request_mbc.json b/cli/tests/lsp/hover_request_mbc.json
index 6821fb0ff..6e7c55e08 100644
--- a/cli/tests/lsp/hover_request_mbc.json
+++ b/cli/tests/lsp/hover_request_mbc.json
@@ -8,7 +8,7 @@
     },
     "position": {
       "line": 2,
-      "character": 14
+      "character": 15
     }
   }
 }