Repository: denoland/deno (language: rust)

Issue:
refactor(lsp): contain language server state to worker thread
Builds on #28035.

Closes #18079.


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 759aebd9117b72..cae534319ca0a3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1430,9 +1430,9 @@ dependencies = [
 
 [[package]]
 name = "deno-tower-lsp-macros"
-version = "0.9.0"
+version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05d59a1cfd445fd86f63616127a434aabca000e03d963b01b03ce813520565b9"
+checksum = "e017686632f7908cde3e02cd1598e25488f57448600a09211834cfd547b9ec39"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2573,9 +2573,9 @@ dependencies = [
 
 [[package]]
 name = "deno_tower_lsp"
-version = "0.1.0"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7afb4d257c084fd889e8cf1ba3ad58db0002428c819cc7717815f996f97777a"
+checksum = "c58d75640e2edffc77929064f2f13c308e549b4b5c42ea6159b0607add856675"
 dependencies = [
  "async-trait",
  "auto_impl",
@@ -2590,7 +2590,7 @@ dependencies = [
  "serde_json",
  "tokio",
  "tokio-util",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
diff --git a/Cargo.toml b/Cargo.toml
index c8fa80ebb34241..4515e8884b3132 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -218,7 +218,7 @@ tokio-socks = "0.5.1"
 tokio-util = "0.7.4"
 tower = { version = "0.5.2", default-features = false, features = ["retry", "util"] }
 tower-http = { version = "0.6.1", features = ["decompression-br", "decompression-gzip"] }
-tower-lsp = { package = "deno_tower_lsp", version = "0.1.0", features = ["proposed"] }
+tower-lsp = { package = "deno_tower_lsp", version = "=0.3.0", features = ["proposed"] }
 tower-service = "0.3.2"
 twox-hash = "=1.6.3"
 url = { version = "2.5", features = ["serde", "expose_internals"] }
diff --git a/cli/bench/lsp.rs b/cli/bench/lsp.rs
index 7a93dcae1eb93e..9896d340f67621 100644
--- a/cli/bench/lsp.rs
+++ b/cli/bench/lsp.rs
@@ -71,7 +71,7 @@ fn patch_uris<'a>(
     };
 
     if let Some(new_req) = new_req {
-      *req = new_req;
+      *req = new_req.request;
     }
   }
 }
diff --git a/cli/jsr.rs b/cli/jsr.rs
index 34a2ec04e4de46..b3b6faf918c891 100644
--- a/cli/jsr.rs
+++ b/cli/jsr.rs
@@ -66,13 +66,11 @@ impl JsrFetchResolver {
     }
     let fetch_package_info = || async {
       let meta_url = jsr_url().join(&format!("{}/meta.json", name)).ok()?;
-      let file_fetcher = self.file_fetcher.clone();
-      // spawn due to the lsp's `Send` requirement
-      let file = deno_core::unsync::spawn(async move {
-        file_fetcher.fetch_bypass_permissions(&meta_url).await.ok()
-      })
-      .await
-      .ok()??;
+      let file = self
+        .file_fetcher
+        .fetch_bypass_permissions(&meta_url)
+        .await
+        .ok()?;
       serde_json::from_slice::<JsrPackageInfo>(&file.source).ok()
     };
     let info = fetch_package_info().await.map(Arc::new);
@@ -92,12 +90,10 @@ impl JsrFetchResolver {
         .join(&format!("{}/{}_meta.json", &nv.name, &nv.version))
         .ok()?;
       let file_fetcher = self.file_fetcher.clone();
-      // spawn due to the lsp's `Send` requirement
-      let file = deno_core::unsync::spawn(async move {
-        file_fetcher.fetch_bypass_permissions(&meta_url).await.ok()
-      })
-      .await
-      .ok()??;
+      let file = file_fetcher
+        .fetch_bypass_permissions(&meta_url)
+        .await
+        .ok()?;
       partial_jsr_package_version_info_from_slice(&file.source).ok()
     };
     let info = fetch_package_version_info().await.map(Arc::new);
diff --git a/cli/lsp/config.rs b/cli/lsp/config.rs
index 2e02d839e4ad19..f0c06996393b3c 100644
--- a/cli/lsp/config.rs
+++ b/cli/lsp/config.rs
@@ -1534,17 +1534,11 @@ impl ConfigData {
           import_map_url.clone(),
           ConfigWatchedFileType::ImportMap,
         );
-        // spawn due to the lsp's `Send` requirement
-        let fetch_result =
-          deno_core::unsync::spawn({
-            let file_fetcher = file_fetcher.cloned().unwrap();
-            let import_map_url = import_map_url.clone();
-            async move {
-              file_fetcher.fetch_bypass_permissions(&import_map_url).await
-            }
-          })
-          .await
-          .unwrap();
+        let fetch_result = file_fetcher
+          .as_ref()
+          .unwrap()
+          .fetch_bypass_permissions(import_map_url)
+          .await;
 
         let value_result = fetch_result.and_then(|f| {
           serde_json::from_slice::<Value>(&f.source).map_err(|e| e.into())
diff --git a/cli/lsp/jsr.rs b/cli/lsp/jsr.rs
index 1bcd6f39304d58..c0c647b0893e00 100644
--- a/cli/lsp/jsr.rs
+++ b/cli/lsp/jsr.rs
@@ -307,7 +307,7 @@ impl CliJsrSearchApi {
   }
 }
 
-#[async_trait::async_trait]
+#[async_trait::async_trait(?Send)]
 impl PackageSearchApi for CliJsrSearchApi {
   async fn search(&self, query: &str) -> Result<Arc<Vec<String>>, AnyError> {
     if let Some(names) = self.search_cache.get(query) {
@@ -316,12 +316,10 @@ impl PackageSearchApi for CliJsrSearchApi {
     let mut search_url = jsr_api_url().join("packages")?;
     search_url.query_pairs_mut().append_pair("query", query);
     let file_fetcher = self.file_fetcher.clone();
-    // spawn due to the lsp's `Send` requirement
-    let file = deno_core::unsync::spawn(async move {
+    let file = {
       let file = file_fetcher.fetch_bypass_permissions(&search_url).await?;
-      TextDecodedFile::decode(file)
-    })
-    .await??;
+      TextDecodedFile::decode(file)?
+    };
     let names = Arc::new(parse_jsr_search_response(&file.source)?);
     self.search_cache.insert(query.to_string(), names.clone());
     Ok(names)
diff --git a/cli/lsp/language_server.rs b/cli/lsp/language_server.rs
index c8d8e83e06ee2a..01cf2d63d2cc7e 100644
--- a/cli/lsp/language_server.rs
+++ b/cli/lsp/language_server.rs
@@ -8,6 +8,7 @@ use std::collections::VecDeque;
 use std::env;
 use std::fmt::Write as _;
 use std::path::PathBuf;
+use std::rc::Rc;
 use std::str::FromStr;
 use std::sync::Arc;
 
@@ -32,7 +33,6 @@ use deno_lib::version::DENO_VERSION_INFO;
 use deno_path_util::url_to_file_path;
 use deno_runtime::deno_tls::rustls::RootCertStore;
 use deno_runtime::deno_tls::RootCertStoreProvider;
-use deno_runtime::tokio_util::create_basic_runtime;
 use deno_semver::jsr::JsrPackageReqReference;
 use indexmap::Equivalent;
 use indexmap::IndexSet;
@@ -129,7 +129,7 @@ impl RootCertStoreProvider for LspRootCertStoreProvider {
 #[derive(Debug, Clone)]
 pub struct LanguageServer {
   client: Client,
-  pub inner: Arc<tokio::sync::RwLock<Inner>>,
+  pub inner: Rc<tokio::sync::RwLock<Inner>>,
   /// This is used to block out standard request handling until the complete
   /// user configuration has been fetched. This is done in the `initialized`
   /// handler which normally may occur concurrently with those other requests.
@@ -139,9 +139,6 @@ pub struct LanguageServer {
   init_flag: AsyncFlag,
   performance: Arc<Performance>,
   shutdown_flag: AsyncFlag,
-  /// This is used to move all request handling to a separate thread so blocking
-  /// code there won't block cancellations from being flagged.
-  worker_thread: Arc<WorkerThread>,
 }
 
 /// Snapshot of the state used by TSC.
@@ -192,73 +189,6 @@ impl LanguageServerTaskQueue {
   }
 }
 
-#[derive(Debug)]
-struct WorkerThread {
-  runtime_handle: tokio::runtime::Handle,
-  shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
-  join_handle: Option<std::thread::JoinHandle<()>>,
-}
-
-impl WorkerThread {
-  fn create() -> Self {
-    let (handle_tx, handle_rx) = std::sync::mpsc::channel();
-    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
-    let join_handle = std::thread::spawn(move || {
-      let runtime = create_basic_runtime();
-      handle_tx.send(runtime.handle().clone()).unwrap();
-      runtime.block_on(shutdown_rx).unwrap();
-    });
-    let runtime_handle = handle_rx.recv().unwrap();
-    Self {
-      runtime_handle,
-      shutdown_tx: Some(shutdown_tx),
-      join_handle: Some(join_handle),
-    }
-  }
-
-  async fn spawn(
-    &self,
-    fut: impl std::future::Future<Output = ()> + Send + 'static,
-  ) {
-    self
-      .runtime_handle
-      .spawn(fut)
-      .await
-      .inspect_err(|err| lsp_warn!("Handler task error: {err}"))
-      .ok();
-  }
-
-  async fn spawn_with_cancellation<
-    T: Send + 'static,
-    TFut: std::future::Future<Output = LspResult<T>> + Send,
-  >(
-    &self,
-    get_fut: impl FnOnce(CancellationToken) -> TFut + Send + 'static,
-  ) -> LspResult<T> {
-    let token = CancellationToken::new();
-    let _drop_guard = token.clone().drop_guard();
-    self
-      .runtime_handle
-      .spawn(async move {
-        tokio::select! {
-          biased;
-          _ = token.cancelled() => Err(LspError::request_cancelled()),
-          result = get_fut(token.clone()) => result,
-        }
-      })
-      .await
-      .inspect_err(|err| lsp_warn!("Handler task error: {err}"))
-      .unwrap_or_else(|_| Err(LspError::internal_error()))
-  }
-}
-
-impl Drop for WorkerThread {
-  fn drop(&mut self) {
-    self.shutdown_tx.take().unwrap().send(()).unwrap();
-    self.join_handle.take().unwrap().join().unwrap();
-  }
-}
-
 #[derive(Debug)]
 pub struct Inner {
   /// Cached versions of "fixed" assets that can either be inlined in Rust or
@@ -304,14 +234,13 @@ impl LanguageServer {
     let performance = Arc::new(Performance::default());
     Self {
       client: client.clone(),
-      inner: Arc::new(tokio::sync::RwLock::new(Inner::new(
+      inner: Rc::new(tokio::sync::RwLock::new(Inner::new(
         client,
         performance.clone(),
       ))),
       init_flag: Default::default(),
       performance,
       shutdown_flag,
-      worker_thread: Arc::new(WorkerThread::create()),
     }
   }
 
@@ -322,22 +251,6 @@ impl LanguageServer {
     specifiers: Vec<ModuleSpecifier>,
     referrer: ModuleSpecifier,
     force_global_cache: bool,
-  ) -> LspResult<Option<Value>> {
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |_| async move {
-        ls.cache_inner(specifiers, referrer, force_global_cache)
-          .await
-      })
-      .await
-  }
-
-  async fn cache_inner(
-    &self,
-    specifiers: Vec<ModuleSpecifier>,
-    referrer: ModuleSpecifier,
-    force_global_cache: bool,
   ) -> LspResult<Option<Value>> {
     async fn create_graph_for_caching(
       factory: CliFactory,
@@ -434,113 +347,88 @@ impl LanguageServer {
   /// coordinate the tests receiving the latest diagnostics.
   pub async fn latest_diagnostic_batch_index_request(
     &self,
+    _token: CancellationToken,
   ) -> LspResult<Option<Value>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |_| async move {
-        Ok(
-          ls.inner
-            .read()
-            .await
-            .diagnostics_server
-            .latest_batch_index()
-            .map(|v| v.into()),
-        )
-      })
-      .await
+    Ok(
+      self
+        .inner
+        .read()
+        .await
+        .diagnostics_server
+        .latest_batch_index()
+        .map(|v| v.into()),
+    )
   }
 
-  pub async fn performance_request(&self) -> LspResult<Option<Value>> {
+  pub async fn performance_request(
+    &self,
+    _token: CancellationToken,
+  ) -> LspResult<Option<Value>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |_| async move {
-        Ok(Some(ls.inner.read().await.get_performance()))
-      })
-      .await
+    Ok(Some(self.inner.read().await.get_performance()))
   }
 
-  pub async fn task_definitions(&self) -> LspResult<Vec<TaskDefinition>> {
+  pub async fn task_definitions(
+    &self,
+    _token: CancellationToken,
+  ) -> LspResult<Vec<TaskDefinition>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |_| async move {
-        ls.inner.read().await.task_definitions()
-      })
-      .await
+    self.inner.read().await.task_definitions()
   }
 
   pub async fn test_run_request(
     &self,
     params: Option<Value>,
+    _token: CancellationToken,
   ) -> LspResult<Option<Value>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |_| async move {
-        ls.inner.read().await.test_run_request(params).await
-      })
-      .await
+    self.inner.read().await.test_run_request(params).await
   }
 
   pub async fn test_run_cancel_request(
     &self,
     params: Option<Value>,
+    _token: CancellationToken,
   ) -> LspResult<Option<Value>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |_| async move {
-        ls.inner.read().await.test_run_cancel_request(params)
-      })
-      .await
+    self.inner.read().await.test_run_cancel_request(params)
   }
 
   pub async fn virtual_text_document(
     &self,
     params: Option<Value>,
+    _token: CancellationToken,
   ) -> LspResult<Option<Value>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |_| async move {
-        match params.map(serde_json::from_value) {
-          Some(Ok(params)) => Ok(Some(
-            serde_json::to_value(
-              ls.inner.read().await.virtual_text_document(params)?,
-            )
-            .map_err(|err| {
-              error!(
-                "Failed to serialize virtual_text_document response: {:#}",
-                err
-              );
-              LspError::internal_error()
-            })?,
-          )),
-          Some(Err(err)) => Err(LspError::invalid_params(err.to_string())),
-          None => Err(LspError::invalid_params("Missing parameters")),
-        }
-      })
-      .await
+    match params.map(serde_json::from_value) {
+      Some(Ok(params)) => Ok(Some(
+        serde_json::to_value(
+          self.inner.read().await.virtual_text_document(params)?,
+        )
+        .map_err(|err| {
+          error!(
+            "Failed to serialize virtual_text_document response: {:#}",
+            err
+          );
+          LspError::internal_error()
+        })?,
+      )),
+      Some(Err(err)) => Err(LspError::invalid_params(err.to_string())),
+      None => Err(LspError::invalid_params("Missing parameters")),
+    }
   }
 
   pub async fn refresh_configuration(&self) {
@@ -1278,13 +1166,10 @@ impl Inner {
 
   async fn refresh_dep_info(&mut self) {
     let dep_info_by_scope = self.documents.dep_info_by_scope();
-    let resolver = self.resolver.clone();
-    // spawn due to the lsp's `Send` requirement
-    spawn(
-      async move { resolver.set_dep_info_by_scope(&dep_info_by_scope).await },
-    )
-    .await
-    .ok();
+    self
+      .resolver
+      .set_dep_info_by_scope(&dep_info_by_scope)
+      .await;
   }
 
   async fn did_close(&mut self, params: DidCloseTextDocumentParams) {
@@ -3526,11 +3411,12 @@ impl Inner {
   }
 }
 
-#[tower_lsp::async_trait]
+#[tower_lsp::async_trait(?Send)]
 impl tower_lsp::LanguageServer for LanguageServer {
   async fn execute_command(
     &self,
     params: ExecuteCommandParams,
+    _token: CancellationToken,
   ) -> LspResult<Option<Value>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
@@ -3620,52 +3506,28 @@ impl tower_lsp::LanguageServer for LanguageServer {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn(async move {
-        ls.inner.write().await.did_open(params).await;
-      })
-      .await;
+    self.inner.write().await.did_open(params).await;
   }
 
   async fn did_change(&self, params: DidChangeTextDocumentParams) {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn(async move {
-        ls.inner.write().await.did_change(params).await;
-      })
-      .await;
+    self.inner.write().await.did_change(params).await;
   }
 
   async fn did_save(&self, params: DidSaveTextDocumentParams) {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn(async move {
-        ls.inner.write().await.did_save(params);
-      })
-      .await;
+    self.inner.write().await.did_save(params);
   }
 
   async fn did_close(&self, params: DidCloseTextDocumentParams) {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn(async move {
-        ls.inner.write().await.did_close(params).await;
-      })
-      .await;
+    self.inner.write().await.did_close(params).await;
   }
 
   async fn did_change_configuration(
@@ -3675,22 +3537,17 @@ impl tower_lsp::LanguageServer for LanguageServer {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
+    let mark = self
+      .performance
+      .mark_with_args("lsp.did_change_configuration", &params);
+    self.refresh_configuration().await;
     self
-      .worker_thread
-      .spawn(async move {
-        let mark = ls
-          .performance
-          .mark_with_args("lsp.did_change_configuration", &params);
-        ls.refresh_configuration().await;
-        ls.inner
-          .write()
-          .await
-          .did_change_configuration(params)
-          .await;
-        ls.performance.measure(mark);
-      })
+      .inner
+      .write()
+      .await
+      .did_change_configuration(params)
       .await;
+    self.performance.measure(mark);
   }
 
   async fn did_change_watched_files(
@@ -3700,16 +3557,11 @@ impl tower_lsp::LanguageServer for LanguageServer {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn(async move {
-        ls.inner
-          .write()
-          .await
-          .did_change_watched_files(params)
-          .await;
-      })
+      .inner
+      .write()
+      .await
+      .did_change_watched_files(params)
       .await;
   }
 
@@ -3720,476 +3572,373 @@ impl tower_lsp::LanguageServer for LanguageServer {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
+    let mark = self
+      .performance
+      .mark_with_args("lsp.did_change_workspace_folders", &params);
     self
-      .worker_thread
-      .spawn(async move {
-        let mark = ls
-          .performance
-          .mark_with_args("lsp.did_change_workspace_folders", &params);
-        ls.inner
-          .write()
-          .await
-          .pre_did_change_workspace_folders(params);
-        ls.refresh_configuration().await;
-        ls.inner
-          .write()
-          .await
-          .post_did_change_workspace_folders()
-          .await;
-        ls.performance.measure(mark);
-      })
+      .inner
+      .write()
+      .await
+      .pre_did_change_workspace_folders(params);
+    self.refresh_configuration().await;
+    self
+      .inner
+      .write()
+      .await
+      .post_did_change_workspace_folders()
       .await;
+    self.performance.measure(mark);
   }
 
   async fn document_symbol(
     &self,
     params: DocumentSymbolParams,
+    token: CancellationToken,
   ) -> LspResult<Option<DocumentSymbolResponse>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.document_symbol(params, &token).await
-      })
+      .inner
+      .read()
+      .await
+      .document_symbol(params, &token)
       .await
   }
 
   async fn formatting(
     &self,
     params: DocumentFormattingParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<TextEdit>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.formatting(params, &token).await
-      })
-      .await
+    self.inner.read().await.formatting(params, &token).await
   }
 
-  async fn hover(&self, params: HoverParams) -> LspResult<Option<Hover>> {
+  async fn hover(
+    &self,
+    params: HoverParams,
+    token: CancellationToken,
+  ) -> LspResult<Option<Hover>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.hover(params, &token).await
-      })
-      .await
+    self.inner.read().await.hover(params, &token).await
   }
 
   async fn inlay_hint(
     &self,
     params: InlayHintParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<InlayHint>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.inlay_hint(params, &token).await
-      })
-      .await
+    self.inner.read().await.inlay_hint(params, &token).await
   }
 
   async fn code_action(
     &self,
     params: CodeActionParams,
+    token: CancellationToken,
   ) -> LspResult<Option<CodeActionResponse>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.code_action(params, &token).await
-      })
-      .await
+    self.inner.read().await.code_action(params, &token).await
   }
 
   async fn code_action_resolve(
     &self,
     params: CodeAction,
+    token: CancellationToken,
   ) -> LspResult<CodeAction> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .code_action_resolve(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .code_action_resolve(params, &token)
       .await
   }
 
   async fn code_lens(
     &self,
     params: CodeLensParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<CodeLens>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.code_lens(params, &token).await
-      })
-      .await
+    self.inner.read().await.code_lens(params, &token).await
   }
 
-  async fn code_lens_resolve(&self, params: CodeLens) -> LspResult<CodeLens> {
+  async fn code_lens_resolve(
+    &self,
+    params: CodeLens,
+    token: CancellationToken,
+  ) -> LspResult<CodeLens> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .code_lens_resolve(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .code_lens_resolve(params, &token)
       .await
   }
 
   async fn document_highlight(
     &self,
     params: DocumentHighlightParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<DocumentHighlight>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .document_highlight(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .document_highlight(params, &token)
       .await
   }
 
   async fn references(
     &self,
     params: ReferenceParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<Location>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.references(params, &token).await
-      })
-      .await
+    self.inner.read().await.references(params, &token).await
   }
 
   async fn goto_definition(
     &self,
     params: GotoDefinitionParams,
+    token: CancellationToken,
   ) -> LspResult<Option<GotoDefinitionResponse>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.goto_definition(params, &token).await
-      })
+      .inner
+      .read()
+      .await
+      .goto_definition(params, &token)
       .await
   }
 
   async fn goto_type_definition(
     &self,
     params: GotoTypeDefinitionParams,
+    token: CancellationToken,
   ) -> LspResult<Option<GotoTypeDefinitionResponse>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .goto_type_definition(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .goto_type_definition(params, &token)
       .await
   }
 
   async fn completion(
     &self,
     params: CompletionParams,
+    token: CancellationToken,
   ) -> LspResult<Option<CompletionResponse>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.completion(params, &token).await
-      })
-      .await
+    self.inner.read().await.completion(params, &token).await
   }
 
   async fn completion_resolve(
     &self,
     params: CompletionItem,
+    token: CancellationToken,
   ) -> LspResult<CompletionItem> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .completion_resolve(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .completion_resolve(params, &token)
       .await
   }
 
   async fn goto_implementation(
     &self,
     params: GotoImplementationParams,
+    token: CancellationToken,
   ) -> LspResult<Option<GotoImplementationResponse>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .goto_implementation(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .goto_implementation(params, &token)
       .await
   }
 
   async fn folding_range(
     &self,
     params: FoldingRangeParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<FoldingRange>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.folding_range(params, &token).await
-      })
-      .await
+    self.inner.read().await.folding_range(params, &token).await
   }
 
   async fn incoming_calls(
     &self,
     params: CallHierarchyIncomingCallsParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<CallHierarchyIncomingCall>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.incoming_calls(params, &token).await
-      })
-      .await
+    self.inner.read().await.incoming_calls(params, &token).await
   }
 
   async fn outgoing_calls(
     &self,
     params: CallHierarchyOutgoingCallsParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<CallHierarchyOutgoingCall>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.outgoing_calls(params, &token).await
-      })
-      .await
+    self.inner.read().await.outgoing_calls(params, &token).await
   }
 
   async fn prepare_call_hierarchy(
     &self,
     params: CallHierarchyPrepareParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<CallHierarchyItem>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .prepare_call_hierarchy(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .prepare_call_hierarchy(params, &token)
       .await
   }
 
   async fn rename(
     &self,
     params: RenameParams,
+    token: CancellationToken,
   ) -> LspResult<Option<WorkspaceEdit>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.rename(params, &token).await
-      })
-      .await
+    self.inner.read().await.rename(params, &token).await
   }
 
   async fn selection_range(
     &self,
     params: SelectionRangeParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<SelectionRange>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.selection_range(params, &token).await
-      })
+      .inner
+      .read()
+      .await
+      .selection_range(params, &token)
       .await
   }
 
   async fn semantic_tokens_full(
     &self,
     params: SemanticTokensParams,
+    token: CancellationToken,
   ) -> LspResult<Option<SemanticTokensResult>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .semantic_tokens_full(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .semantic_tokens_full(params, &token)
       .await
   }
 
   async fn semantic_tokens_range(
     &self,
     params: SemanticTokensRangeParams,
+    token: CancellationToken,
   ) -> LspResult<Option<SemanticTokensRangeResult>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .semantic_tokens_range(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .semantic_tokens_range(params, &token)
       .await
   }
 
   async fn signature_help(
     &self,
     params: SignatureHelpParams,
+    token: CancellationToken,
   ) -> LspResult<Option<SignatureHelp>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.signature_help(params, &token).await
-      })
-      .await
+    self.inner.read().await.signature_help(params, &token).await
   }
 
   async fn will_rename_files(
     &self,
     params: RenameFilesParams,
+    token: CancellationToken,
   ) -> LspResult<Option<WorkspaceEdit>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
     self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner
-          .read()
-          .await
-          .will_rename_files(params, &token)
-          .await
-      })
+      .inner
+      .read()
+      .await
+      .will_rename_files(params, &token)
       .await
   }
 
   async fn symbol(
     &self,
     params: WorkspaceSymbolParams,
+    token: CancellationToken,
   ) -> LspResult<Option<Vec<SymbolInformation>>> {
     if !self.init_flag.is_raised() {
       self.init_flag.wait_raised().await;
     }
-    let ls = self.clone();
-    self
-      .worker_thread
-      .spawn_with_cancellation(move |token| async move {
-        ls.inner.read().await.symbol(params, &token).await
-      })
-      .await
+    self.inner.read().await.symbol(params, &token).await
   }
 }
 
diff --git a/cli/lsp/mod.rs b/cli/lsp/mod.rs
index 9f11cc69900286..79af61c9e77a7c 100644
--- a/cli/lsp/mod.rs
+++ b/cli/lsp/mod.rs
@@ -74,7 +74,7 @@ pub async fn start() -> Result<(), AnyError> {
     builder
   };
 
-  let (service, socket) = builder.finish();
+  let (service, socket, pending) = builder.finish();
 
   // TODO(nayeemrmn): This shutdown flag is a workaround for
   // https://github.com/denoland/deno/issues/20700. Remove when
@@ -82,7 +82,7 @@ pub async fn start() -> Result<(), AnyError> {
   // Force end the server 8 seconds after receiving a shutdown request.
   tokio::select! {
     biased;
-    _ = Server::new(stdin, stdout, socket).concurrency_level(32).serve(service) => {}
+    _ = Server::new(stdin, stdout, socket, pending).concurrency_level(32).serve(service) => {}
     _ = spawn(async move {
       shutdown_flag.wait_raised().await;
       tokio::time::sleep(std::time::Duration::from_secs(8)).await;
diff --git a/cli/lsp/npm.rs b/cli/lsp/npm.rs
index 20e48ba49b88ac..e58df36e2466a2 100644
--- a/cli/lsp/npm.rs
+++ b/cli/lsp/npm.rs
@@ -48,7 +48,7 @@ impl CliNpmSearchApi {
   }
 }
 
-#[async_trait::async_trait]
+#[async_trait::async_trait(?Send)]
 impl PackageSearchApi for CliNpmSearchApi {
   async fn search(&self, query: &str) -> Result<Arc<Vec<String>>, AnyError> {
     if let Some(names) = self.search_cache.get(query) {
diff --git a/cli/lsp/performance.rs b/cli/lsp/performance.rs
index 8d836c7adfb76c..cf7d4dee0a4a7a 100644
--- a/cli/lsp/performance.rs
+++ b/cli/lsp/performance.rs
@@ -74,14 +74,15 @@ impl From<PerformanceMark> for PerformanceMeasure {
 
 #[derive(Debug)]
 pub struct PerformanceScopeMark {
-  performance_inner: Arc<Mutex<PerformanceInner>>,
+  performance: Arc<Performance>,
   inner: Option<PerformanceMark>,
 }
 
 impl Drop for PerformanceScopeMark {
   fn drop(&mut self) {
     self
-      .performance_inner
+      .performance
+      .0
       .lock()
       .measure(self.inner.take().unwrap());
   }
@@ -139,7 +140,7 @@ impl Default for PerformanceInner {
 /// The structure will limit the size of measurements to the most recent 1000,
 /// and will roll off when that limit is reached.
 #[derive(Debug, Default)]
-pub struct Performance(Arc<Mutex<PerformanceInner>>);
+pub struct Performance(Mutex<PerformanceInner>);
 
 impl Performance {
   /// Return the count and average duration of a measurement identified by name.
@@ -279,9 +280,12 @@ impl Performance {
   /// // ❌
   /// let _ = self.performance.measure_scope("foo");
   /// ```
-  pub fn measure_scope<S: AsRef<str>>(&self, name: S) -> PerformanceScopeMark {
+  pub fn measure_scope<S: AsRef<str>>(
+    self: &Arc<Self>,
+    name: S,
+  ) -> PerformanceScopeMark {
     PerformanceScopeMark {
-      performance_inner: self.0.clone(),
+      performance: self.clone(),
       inner: Some(self.mark(name)),
     }
   }
diff --git a/cli/lsp/registries.rs b/cli/lsp/registries.rs
index f8c38d97ea46a6..701aaf131ab7de 100644
--- a/cli/lsp/registries.rs
+++ b/cli/lsp/registries.rs
@@ -474,24 +474,17 @@ impl ModuleRegistry {
     &self,
     specifier: &ModuleSpecifier,
   ) -> Result<Vec<RegistryConfiguration>, AnyError> {
-    // spawn due to the lsp's `Send` requirement
-    let fetch_result = deno_core::unsync::spawn({
-      let file_fetcher = self.file_fetcher.clone();
-      let specifier = specifier.clone();
-      async move {
-        file_fetcher
-        .fetch_with_options(
-          &specifier,
-FetchPermissionsOptionRef::AllowAll,
-          FetchOptions {
-            maybe_auth: None,
-            maybe_accept: Some("application/vnd.deno.reg.v2+json, application/vnd.deno.reg.v1+json;q=0.9, application/json;q=0.8"),
-            maybe_cache_setting: None,
-          }
-        )
-        .await
-      }
-    }).await?;
+    let fetch_result = self.file_fetcher
+      .fetch_with_options(
+        specifier,
+        FetchPermissionsOptionRef::AllowAll,
+        FetchOptions {
+          maybe_auth: None,
+          maybe_accept: Some("application/vnd.deno.reg.v2+json, application/vnd.deno.reg.v1+json;q=0.9, application/json;q=0.8"),
+          maybe_cache_setting: None,
+        }
+      )
+      .await;
     // if there is an error fetching, we will cache an empty file, so that
     // subsequent requests they are just an empty doc which will error without
     // needing to connect to the remote URL. We will cache it for 1 week.
@@ -584,18 +577,13 @@ FetchPermissionsOptionRef::AllowAll,
         )
         .ok()?;
         let file_fetcher = self.file_fetcher.clone();
-        // spawn due to the lsp's `Send` requirement
-        let file = deno_core::unsync::spawn({
-          async move {
-            let file = file_fetcher
-              .fetch_bypass_permissions(&endpoint)
-              .await
-              .ok()?;
-            TextDecodedFile::decode(file).ok()
-          }
-        })
-        .await
-        .ok()??;
+        let file = {
+          let file = file_fetcher
+            .fetch_bypass_permissions(&endpoint)
+            .await
+            .ok()?;
+          TextDecodedFile::decode(file).ok()?
+        };
         let documentation: lsp::Documentation =
           serde_json::from_str(&file.source).ok()?;
         return match documentation {
@@ -983,16 +971,13 @@ FetchPermissionsOptionRef::AllowAll,
   ) -> Option<lsp::Documentation> {
     let specifier = Url::parse(url).ok()?;
     let file_fetcher = self.file_fetcher.clone();
-    // spawn due to the lsp's `Send` requirement
-    let file = deno_core::unsync::spawn(async move {
+    let file = {
       let file = file_fetcher
         .fetch_bypass_permissions(&specifier)
         .await
         .ok()?;
-      TextDecodedFile::decode(file).ok()
-    })
-    .await
-    .ok()??;
+      TextDecodedFile::decode(file).ok()?
+    };
     serde_json::from_str(&file.source).ok()
   }
 
@@ -1045,26 +1030,20 @@ FetchPermissionsOptionRef::AllowAll,
 
   async fn get_items(&self, url: &str) -> Option<VariableItems> {
     let specifier = ModuleSpecifier::parse(url).ok()?;
-    // spawn due to the lsp's `Send` requirement
-    let file = deno_core::unsync::spawn({
-      let file_fetcher = self.file_fetcher.clone();
-      let specifier = specifier.clone();
-      async move {
-        let file = file_fetcher
-          .fetch_bypass_permissions(&specifier)
-          .await
-          .map_err(|err| {
-            error!(
-              "Internal error fetching endpoint \"{}\". {}",
-              specifier, err
-            );
-          })
-          .ok()?;
-        TextDecodedFile::decode(file).ok()
-      }
-    })
-    .await
-    .ok()??;
+    let file = {
+      let file = self
+        .file_fetcher
+        .fetch_bypass_permissions(&specifier)
+        .await
+        .map_err(|err| {
+          error!(
+            "Internal error fetching endpoint \"{}\". {}",
+            specifier, err
+          );
+        })
+        .ok()?;
+      TextDecodedFile::decode(file).ok()?
+    };
     let items: VariableItems = serde_json::from_str(&file.source)
       .map_err(|err| {
         error!(
@@ -1090,26 +1069,20 @@ FetchPermissionsOptionRef::AllowAll,
           error!("Internal error mapping endpoint \"{}\". {}", url, err);
         })
         .ok()?;
-    // spawn due to the lsp's `Send` requirement
-    let file = deno_core::unsync::spawn({
-      let file_fetcher = self.file_fetcher.clone();
-      let specifier = specifier.clone();
-      async move {
-        let file = file_fetcher
-          .fetch_bypass_permissions(&specifier)
-          .await
-          .map_err(|err| {
-            error!(
-              "Internal error fetching endpoint \"{}\". {}",
-              specifier, err
-            );
-          })
-          .ok()?;
-        TextDecodedFile::decode(file).ok()
-      }
-    })
-    .await
-    .ok()??;
+    let file = {
+      let file = self
+        .file_fetcher
+        .fetch_bypass_permissions(&specifier)
+        .await
+        .map_err(|err| {
+          error!(
+            "Internal error fetching endpoint \"{}\". {}",
+            specifier, err
+          );
+        })
+        .ok()?;
+      TextDecodedFile::decode(file).ok()?
+    };
     let items: VariableItems = serde_json::from_str(&file.source)
       .map_err(|err| {
         error!(
diff --git a/cli/lsp/repl.rs b/cli/lsp/repl.rs
index a30427312d3741..f7a20c0253049c 100644
--- a/cli/lsp/repl.rs
+++ b/cli/lsp/repl.rs
@@ -9,6 +9,7 @@ use deno_core::anyhow::anyhow;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
 use lsp_types::Uri;
+use tokio_util::sync::CancellationToken;
 use tower_lsp::lsp_types::ClientCapabilities;
 use tower_lsp::lsp_types::ClientInfo;
 use tower_lsp::lsp_types::CompletionContext;
@@ -124,6 +125,7 @@ impl ReplLanguageServer {
     &mut self,
     line_text: &str,
     position: usize,
+    token: CancellationToken,
   ) -> Vec<ReplCompletionItem> {
     self.did_change(line_text).await;
     let text_info = deno_ast::SourceTextInfo::from_string(format!(
@@ -135,27 +137,30 @@ impl ReplLanguageServer {
     let line_and_column = text_info.line_and_column_index(position);
     let response = self
       .language_server
-      .completion(CompletionParams {
-        text_document_position: TextDocumentPositionParams {
-          text_document: TextDocumentIdentifier {
-            uri: self.get_document_uri(),
+      .completion(
+        CompletionParams {
+          text_document_position: TextDocumentPositionParams {
+            text_document: TextDocumentIdentifier {
+              uri: self.get_document_uri(),
+            },
+            position: Position {
+              line: line_and_column.line_index as u32,
+              character: line_and_column.column_index as u32,
+            },
           },
-          position: Position {
-            line: line_and_column.line_index as u32,
-            character: line_and_column.column_index as u32,
+          work_done_progress_params: WorkDoneProgressParams {
+            work_done_token: None,
           },
+          partial_result_params: PartialResultParams {
+            partial_result_token: None,
+          },
+          context: Some(CompletionContext {
+            trigger_kind: CompletionTriggerKind::INVOKED,
+            trigger_character: None,
+          }),
         },
-        work_done_progress_params: WorkDoneProgressParams {
-          work_done_token: None,
-        },
-        partial_result_params: PartialResultParams {
-          partial_result_token: None,
-        },
-        context: Some(CompletionContext {
-          trigger_kind: CompletionTriggerKind::INVOKED,
-          trigger_character: None,
-        }),
-      })
+        token,
+      )
       .await
       .ok()
       .unwrap_or_default();
diff --git a/cli/lsp/resolver.rs b/cli/lsp/resolver.rs
index 074372aab32994..5aa17d2b012f4c 100644
--- a/cli/lsp/resolver.rs
+++ b/cli/lsp/resolver.rs
@@ -796,15 +796,9 @@ impl<'a> ResolverFactory<'a> {
         NpmSystemInfo::default(),
       ));
       self.set_npm_installer(npm_installer);
-      // spawn due to the lsp's `Send` requirement
-      deno_core::unsync::spawn(async move {
-        if let Err(err) = npm_resolution_initializer.ensure_initialized().await
-        {
-          log::warn!("failed to initialize npm resolution: {}", err);
-        }
-      })
-      .await
-      .unwrap();
+      if let Err(err) = npm_resolution_initializer.ensure_initialized().await {
+        log::warn!("failed to initialize npm resolution: {}", err);
+      }
 
       CliNpmResolverCreateOptions::Managed(CliManagedNpmResolverCreateOptions {
         sys: CliSys::default(),
diff --git a/cli/lsp/search.rs b/cli/lsp/search.rs
index 281542742d42f6..f3811b47ea3894 100644
--- a/cli/lsp/search.rs
+++ b/cli/lsp/search.rs
@@ -6,7 +6,7 @@ use deno_core::error::AnyError;
 use deno_semver::package::PackageNv;
 use deno_semver::Version;
 
-#[async_trait::async_trait]
+#[async_trait::async_trait(?Send)]
 pub trait PackageSearchApi {
   async fn search(&self, query: &str) -> Result<Arc<Vec<String>>, AnyError>;
   async fn versions(&self, name: &str) -> Result<Arc<Vec<Version>>, AnyError>;
@@ -45,7 +45,7 @@ pub mod tests {
     }
   }
 
-  #[async_trait::async_trait]
+  #[async_trait::async_trait(?Send)]
   impl PackageSearchApi for TestPackageSearchApi {
     async fn search(&self, query: &str) -> Result<Arc<Vec<String>>, AnyError> {
       let names = self
diff --git a/cli/npm/mod.rs b/cli/npm/mod.rs
index 388b007f811f8c..be08ffde66d1fc 100644
--- a/cli/npm/mod.rs
+++ b/cli/npm/mod.rs
@@ -152,24 +152,16 @@ impl NpmFetchResolver {
     // todo(#27198): use RegistryInfoProvider instead
     let fetch_package_info = || async {
       let info_url = deno_npm_cache::get_package_url(&self.npmrc, name);
-      let file_fetcher = self.file_fetcher.clone();
       let registry_config = self.npmrc.get_registry_config(name);
       // TODO(bartlomieju): this should error out, not use `.ok()`.
       let maybe_auth_header =
         deno_npm_cache::maybe_auth_header_for_npm_registry(registry_config)
           .ok()?;
-      // spawn due to the lsp's `Send` requirement
-      let file = deno_core::unsync::spawn(async move {
-        file_fetcher
-          .fetch_bypass_permissions_with_maybe_auth(
-            &info_url,
-            maybe_auth_header,
-          )
-          .await
-          .ok()
-      })
-      .await
-      .ok()??;
+      let file = self
+        .file_fetcher
+        .fetch_bypass_permissions_with_maybe_auth(&info_url, maybe_auth_header)
+        .await
+        .ok()?;
       serde_json::from_slice::<NpmPackageInfo>(&file.source).ok()
     };
     let info = fetch_package_info().await.map(Arc::new);
diff --git a/cli/tools/jupyter/mod.rs b/cli/tools/jupyter/mod.rs
index 211f3b90ee670b..e692510efe3c39 100644
--- a/cli/tools/jupyter/mod.rs
+++ b/cli/tools/jupyter/mod.rs
@@ -22,6 +22,7 @@ use jupyter_runtime::messaging::StreamContent;
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::UnboundedSender;
 use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
 
 use crate::args::Flags;
 use crate::args::JupyterFlags;
@@ -390,7 +391,9 @@ impl JupyterReplSession {
         line_text,
         position,
       } => JupyterReplResponse::LspCompletions(
-        self.lsp_completions(&line_text, position).await,
+        self
+          .lsp_completions(&line_text, position, CancellationToken::new())
+          .await,
       ),
       JupyterReplRequest::JsGetProperties { object_id } => {
         JupyterReplResponse::JsGetProperties(
@@ -432,11 +435,12 @@ impl JupyterReplSession {
     &mut self,
     line_text: &str,
     position: usize,
+    token: CancellationToken,
   ) -> Vec<ReplCompletionItem> {
     self
       .repl_session
       .language_server
-      .completions(line_text, position)
+      .completions(line_text, position, token)
       .await
   }
 
diff --git a/cli/tools/repl/mod.rs b/cli/tools/repl/mod.rs
index 579c24fcd4a2d5..a3328ac2a3b553 100644
--- a/cli/tools/repl/mod.rs
+++ b/cli/tools/repl/mod.rs
@@ -11,6 +11,7 @@ use deno_core::unsync::spawn_blocking;
 use deno_lib::version::DENO_VERSION_INFO;
 use deno_runtime::WorkerExecutionMode;
 use rustyline::error::ReadlineError;
+use tokio_util::sync::CancellationToken;
 
 use crate::args::CliOptions;
 use crate::args::Flags;
@@ -118,7 +119,11 @@ async fn read_line_and_poll(
             line_text,
             position,
           }) => {
-            let result = repl_session.language_server.completions(&line_text, position).await;
+            let result = repl_session.language_server.completions(
+              &line_text,
+              position,
+              CancellationToken::new(),
+            ).await;
             message_handler.send(RustylineSyncResponse::LspCompletions(result)).unwrap();
           }
           None => {}, // channel closed


Test patch:
diff --git a/tests/integration/lsp_tests.rs b/tests/integration/lsp_tests.rs
index b526004a558a85..73f12e0242b167 100644
--- a/tests/integration/lsp_tests.rs
+++ b/tests/integration/lsp_tests.rs
@@ -14268,7 +14268,8 @@ fn lsp_node_modules_dir() {
       "unstable": [],
     } }))
   };
-  let diagnostics = refresh_config(&mut client);
+  refresh_config(&mut client);
+  let diagnostics = client.read_diagnostics();
   assert_eq!(diagnostics.all().len(), 2, "{:#?}", diagnostics); // not cached
 
   cache(&mut client);
@@ -14361,7 +14362,7 @@ fn lsp_vendor_dir() {
     temp_dir.path().join("deno.json"),
     "{ \"vendor\": true, \"lock\": false }\n",
   );
-  let diagnostics = client.change_configuration(json!({ "deno": {
+  client.change_configuration(json!({ "deno": {
     "enable": true,
     "config": "./deno.json",
     "codeLens": {
@@ -14379,6 +14380,7 @@ fn lsp_vendor_dir() {
     },
     "unstable": [],
   } }));
+  let diagnostics = client.read_diagnostics();
 
   // won't be cached until a manual cache occurs
   assert_eq!(
diff --git a/tests/util/server/src/lsp.rs b/tests/util/server/src/lsp.rs
index 2e9e378c7bdab2..1d7a7f8cc3ec7f 100644
--- a/tests/util/server/src/lsp.rs
+++ b/tests/util/server/src/lsp.rs
@@ -919,10 +919,7 @@ impl LspClient {
     self.write_notification("textDocument/didOpen", params);
   }
 
-  pub fn change_configuration(
-    &mut self,
-    config: Value,
-  ) -> CollectedDiagnostics {
+  pub fn change_configuration(&mut self, config: Value) {
     self.config = config;
     if self.supports_workspace_configuration {
       self.write_notification(
@@ -936,7 +933,6 @@ impl LspClient {
         json!({ "settings": &self.config }),
       );
     }
-    self.read_diagnostics()
   }
 
   pub fn handle_configuration_request(&mut self) {