diff --git a/Cargo.lock b/Cargo.lock
index cde704c7b0..448ab935aa 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4456,6 +4456,7 @@ dependencies = [
  "hyper",
  "lazy_static",
  "os_pipe",
+ "parking_lot 0.11.2",
  "pretty_assertions",
  "pty",
  "regex",
diff --git a/cli/tests/integration/lsp_tests.rs b/cli/tests/integration/lsp_tests.rs
index 784dc82630..415f0610e8 100644
--- a/cli/tests/integration/lsp_tests.rs
+++ b/cli/tests/integration/lsp_tests.rs
@@ -2188,149 +2188,6 @@ fn lsp_call_hierarchy() {
   shutdown(&mut client);
 }
 
-#[test]
-fn lsp_format_mbc() {
-  let mut client = init("initialize_params.json");
-  did_open(
-    &mut client,
-    json!({
-      "textDocument": {
-        "uri": "file:///a/file.ts",
-        "languageId": "typescript",
-        "version": 1,
-        "text": "const bar = '👍🇺🇸😃'\nconsole.log('hello deno')\n"
-      }
-    }),
-  );
-  let (maybe_res, maybe_err) = client
-    .write_request(
-      "textDocument/formatting",
-      json!({
-        "textDocument": {
-          "uri": "file:///a/file.ts"
-        },
-        "options": {
-          "tabSize": 2,
-          "insertSpaces": true
-        }
-      }),
-    )
-    .unwrap();
-  assert!(maybe_err.is_none());
-  assert_eq!(
-    maybe_res,
-    Some(json!(load_fixture("formatting_mbc_response.json")))
-  );
-  shutdown(&mut client);
-}
-
-#[test]
-fn lsp_format_exclude_with_config() {
-  let temp_dir = TempDir::new().unwrap();
-  let mut params: lsp::InitializeParams =
-    serde_json::from_value(load_fixture("initialize_params.json")).unwrap();
-  let deno_fmt_jsonc =
-    serde_json::to_vec_pretty(&load_fixture("deno.fmt.exclude.jsonc")).unwrap();
-  fs::write(temp_dir.path().join("deno.fmt.jsonc"), deno_fmt_jsonc).unwrap();
-
-  params.root_uri = Some(Url::from_file_path(temp_dir.path()).unwrap());
-  if let Some(Value::Object(mut map)) = params.initialization_options {
-    map.insert("config".to_string(), json!("./deno.fmt.jsonc"));
-    params.initialization_options = Some(Value::Object(map));
-  }
-
-  let deno_exe = deno_exe_path();
-  let mut client = LspClient::new(&deno_exe, false).unwrap();
-  client
-    .write_request::<_, _, Value>("initialize", params)
-    .unwrap();
-
-  let file_uri =
-    ModuleSpecifier::from_file_path(temp_dir.path().join("ignored.ts"))
-      .unwrap()
-      .to_string();
-  did_open(
-    &mut client,
-    json!({
-      "textDocument": {
-        "uri": file_uri,
-        "languageId": "typescript",
-        "version": 1,
-        "text": "function   myFunc(){}"
-      }
-    }),
-  );
-  let (maybe_res, maybe_err) = client
-    .write_request(
-      "textDocument/formatting",
-      json!({
-        "textDocument": {
-          "uri": file_uri
-        },
-        "options": {
-          "tabSize": 2,
-          "insertSpaces": true
-        }
-      }),
-    )
-    .unwrap();
-  assert!(maybe_err.is_none());
-  assert_eq!(maybe_res, Some(json!(null)));
-  shutdown(&mut client);
-}
-
-#[test]
-fn lsp_format_exclude_default_config() {
-  let temp_dir = TempDir::new().unwrap();
-  let workspace_root = temp_dir.path().canonicalize().unwrap();
-  let mut params: lsp::InitializeParams =
-    serde_json::from_value(load_fixture("initialize_params.json")).unwrap();
-  let deno_jsonc =
-    serde_json::to_vec_pretty(&load_fixture("deno.fmt.exclude.jsonc")).unwrap();
-  fs::write(workspace_root.join("deno.jsonc"), deno_jsonc).unwrap();
-
-  params.root_uri = Some(Url::from_file_path(workspace_root.clone()).unwrap());
-
-  let deno_exe = deno_exe_path();
-  let mut client = LspClient::new(&deno_exe, false).unwrap();
-  client
-    .write_request::<_, _, Value>("initialize", params)
-    .unwrap();
-
-  let file_uri =
-    ModuleSpecifier::from_file_path(workspace_root.join("ignored.ts"))
-      .unwrap()
-      .to_string();
-  did_open(
-    &mut client,
-    json!({
-      "textDocument": {
-        "uri": file_uri,
-        "languageId": "typescript",
-        "version": 1,
-        "text": "function   myFunc(){}"
-      }
-    }),
-  );
-  let (maybe_res, maybe_err) = client
-    .write_request(
-      "textDocument/formatting",
-      json!({
-        "textDocument": {
-          "uri": file_uri
-        },
-        "options": {
-          "tabSize": 2,
-          "insertSpaces": true
-        }
-      }),
-    )
-    .unwrap();
-  assert!(maybe_err.is_none());
-  assert_eq!(maybe_res, Some(json!(null)));
-  shutdown(&mut client);
-}
-
 #[test]
 fn lsp_large_doc_changes() {
   let mut client = init("initialize_params.json");
@@ -4433,6 +4290,216 @@ fn lsp_performance() {
   shutdown(&mut client);
 }
 
+#[test]
+fn lsp_format_no_changes() {
+  let mut client = init("initialize_params.json");
+  did_open(
+    &mut client,
+    json!({
+      "textDocument": {
+        "uri": "file:///a/file.ts",
+        "languageId": "typescript",
+        "version": 1,
+        "text": "console;\n"
+      }
+    }),
+  );
+  let (maybe_res, maybe_err) = client
+    .write_request(
+      "textDocument/formatting",
+      json!({
+        "textDocument": {
+          "uri": "file:///a/file.ts"
+        },
+        "options": {
+          "tabSize": 2,
+          "insertSpaces": true
+        }
+      }),
+    )
+    .unwrap();
+  assert!(maybe_err.is_none());
+  assert_eq!(maybe_res, Some(json!(null)));
+  client.assert_no_notification("window/showMessage");
+  shutdown(&mut client);
+}
+
+#[test]
+fn lsp_format_error() {
+  let mut client = init("initialize_params.json");
+  did_open(
+    &mut client,
+    json!({
+      "textDocument": {
+        "uri": "file:///a/file.ts",
+        "languageId": "typescript",
+        "version": 1,
+        "text": "console test test\n"
+      }
+    }),
+  );
+  let (maybe_res, maybe_err) = client
+    .write_request(
+      "textDocument/formatting",
+      json!({
+        "textDocument": {
+          "uri": "file:///a/file.ts"
+        },
+        "options": {
+          "tabSize": 2,
+          "insertSpaces": true
+        }
+      }),
+    )
+    .unwrap();
+  assert!(maybe_err.is_none());
+  assert_eq!(maybe_res, Some(json!(null)));
+  shutdown(&mut client);
+}
+
+#[test]
+fn lsp_format_mbc() {
+  let mut client = init("initialize_params.json");
+  did_open(
+    &mut client,
+    json!({
+      "textDocument": {
+        "uri": "file:///a/file.ts",
+        "languageId": "typescript",
+        "version": 1,
+        "text": "const bar = '👍🇺🇸😃'\nconsole.log('hello deno')\n"
+      }
+    }),
+  );
+  let (maybe_res, maybe_err) = client
+    .write_request(
+      "textDocument/formatting",
+      json!({
+        "textDocument": {
+          "uri": "file:///a/file.ts"
+        },
+        "options": {
+          "tabSize": 2,
+          "insertSpaces": true
+        }
+      }),
+    )
+    .unwrap();
+  assert!(maybe_err.is_none());
+  assert_eq!(
+    maybe_res,
+    Some(json!(load_fixture("formatting_mbc_response.json")))
+  );
+  shutdown(&mut client);
+}
+
+#[test]
+fn lsp_format_exclude_with_config() {
+  let temp_dir = TempDir::new().unwrap();
+  let mut params: lsp::InitializeParams =
+    serde_json::from_value(load_fixture("initialize_params.json")).unwrap();
+  let deno_fmt_jsonc =
+    serde_json::to_vec_pretty(&load_fixture("deno.fmt.exclude.jsonc")).unwrap();
+  fs::write(temp_dir.path().join("deno.fmt.jsonc"), deno_fmt_jsonc).unwrap();
+
+  params.root_uri = Some(Url::from_file_path(temp_dir.path()).unwrap());
+  if let Some(Value::Object(mut map)) = params.initialization_options {
+    map.insert("config".to_string(), json!("./deno.fmt.jsonc"));
+    params.initialization_options = Some(Value::Object(map));
+  }
+
+  let deno_exe = deno_exe_path();
+  let mut client = LspClient::new(&deno_exe, false).unwrap();
+  client
+    .write_request::<_, _, Value>("initialize", params)
+    .unwrap();
+
+  let file_uri =
+    ModuleSpecifier::from_file_path(temp_dir.path().join("ignored.ts"))
+      .unwrap()
+      .to_string();
+  did_open(
+    &mut client,
+    json!({
+      "textDocument": {
+        "uri": file_uri,
+        "languageId": "typescript",
+        "version": 1,
+        "text": "function   myFunc(){}"
+      }
+    }),
+  );
+  let (maybe_res, maybe_err) = client
+    .write_request(
+      "textDocument/formatting",
+      json!({
+        "textDocument": {
+          "uri": file_uri
+        },
+        "options": {
+          "tabSize": 2,
+          "insertSpaces": true
+        }
+      }),
+    )
+    .unwrap();
+  assert!(maybe_err.is_none());
+  assert_eq!(maybe_res, Some(json!(null)));
+  shutdown(&mut client);
+}
+
+#[test]
+fn lsp_format_exclude_default_config() {
+  let temp_dir = TempDir::new().unwrap();
+  let workspace_root = temp_dir.path().canonicalize().unwrap();
+  let mut params: lsp::InitializeParams =
+    serde_json::from_value(load_fixture("initialize_params.json")).unwrap();
+  let deno_jsonc =
+    serde_json::to_vec_pretty(&load_fixture("deno.fmt.exclude.jsonc")).unwrap();
+  fs::write(workspace_root.join("deno.jsonc"), deno_jsonc).unwrap();
+
+  params.root_uri = Some(Url::from_file_path(workspace_root.clone()).unwrap());
+
+  let deno_exe = deno_exe_path();
+  let mut client = LspClient::new(&deno_exe, false).unwrap();
+  client
+    .write_request::<_, _, Value>("initialize", params)
+    .unwrap();
+
+  let file_uri =
+    ModuleSpecifier::from_file_path(workspace_root.join("ignored.ts"))
+      .unwrap()
+      .to_string();
+  did_open(
+    &mut client,
+    json!({
+      "textDocument": {
+        "uri": file_uri,
+        "languageId": "typescript",
+        "version": 1,
+        "text": "function   myFunc(){}"
+      }
+    }),
+  );
+  let (maybe_res, maybe_err) = client
+    .write_request(
+      "textDocument/formatting",
+      json!({
+        "textDocument": {
+          "uri": file_uri
+        },
+        "options": {
+          "tabSize": 2,
+          "insertSpaces": true
+        }
+      }),
+    )
+    .unwrap();
+  assert!(maybe_err.is_none());
+  assert_eq!(maybe_res, Some(json!(null)));
+  shutdown(&mut client);
+}
+
 #[test]
 fn lsp_format_json() {
   let mut client = init("initialize_params.json");
diff --git a/test_util/Cargo.toml b/test_util/Cargo.toml
index db42c08566..50ccc8decf 100644
--- a/test_util/Cargo.toml
+++ b/test_util/Cargo.toml
@@ -20,6 +20,7 @@ futures = "0.3.21"
 hyper = { version = "0.14.12", features = ["server", "http1", "http2", "runtime"] }
 lazy_static = "1.4.0"
 os_pipe = "1.0.1"
+parking_lot = "0.11.1"
 pretty_assertions = "=1.2.0"
 regex = "1.5.5"
 rustls-pemfile = "0.2.1"
diff --git a/test_util/src/lsp.rs b/test_util/src/lsp.rs
index 9d5a74eafc..c898856bf5 100644
--- a/test_util/src/lsp.rs
+++ b/test_util/src/lsp.rs
@@ -4,13 +4,14 @@ use super::new_deno_dir;
 
 use anyhow::Result;
 use lazy_static::lazy_static;
+use parking_lot::Condvar;
+use parking_lot::Mutex;
 use regex::Regex;
 use serde::de;
 use serde::Deserialize;
 use serde::Serialize;
 use serde_json::json;
 use serde_json::Value;
-use std::collections::VecDeque;
 use std::io;
 use std::io::Write;
 use std::path::Path;
@@ -19,6 +20,7 @@ use std::process::ChildStdin;
 use std::process::ChildStdout;
 use std::process::Command;
 use std::process::Stdio;
+use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
 use tempfile::TempDir;
@@ -28,14 +30,14 @@ lazy_static! {
     Regex::new(r"(?i)^content-length:\s+(\d+)").unwrap();
 }
 
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct LspResponseError {
   code: i32,
   message: String,
   data: Option<Value>,
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub enum LspMessage {
   Notification(String, Option<Value>),
   Request(u64, String, Option<Value>),
@@ -64,14 +66,16 @@ impl<'a> From<&'a [u8]> for LspMessage {
   }
 }
 
-fn read_message<R>(reader: &mut R) -> Result<Vec<u8>>
+fn read_message<R>(reader: &mut R) -> Result<Option<Vec<u8>>>
 where
   R: io::Read + io::BufRead,
 {
   let mut content_length = 0_usize;
   loop {
     let mut buf = String::new();
-    reader.read_line(&mut buf)?;
+    if reader.read_line(&mut buf)? == 0 {
+      return Ok(None);
+    }
     if let Some(captures) = CONTENT_TYPE_REG.captures(&buf) {
       let content_length_match = captures
         .get(1)
@@ -85,16 +89,70 @@ where
 
   let mut msg_buf = vec![0_u8; content_length];
   reader.read_exact(&mut msg_buf)?;
-  Ok(msg_buf)
+  Ok(Some(msg_buf))
+}
+
+struct LspStdoutReader {
+  pending_messages: Arc<(Mutex<Vec<LspMessage>>, Condvar)>,
+  read_messages: Vec<LspMessage>,
+}
+
+impl LspStdoutReader {
+  pub fn new(mut buf_reader: io::BufReader<ChildStdout>) -> Self {
+    let messages: Arc<(Mutex<Vec<LspMessage>>, Condvar)> = Default::default();
+    std::thread::spawn({
+      let messages = messages.clone();
+      move || {
+        while let Ok(Some(msg_buf)) = read_message(&mut buf_reader) {
+          let msg = LspMessage::from(msg_buf.as_slice());
+          let cvar = &messages.1;
+          {
+            let mut messages = messages.0.lock();
+            messages.push(msg);
+          }
+          cvar.notify_all();
+        }
+      }
+    });
+
+    LspStdoutReader {
+      pending_messages: messages,
+      read_messages: Vec::new(),
+    }
+  }
+
+  pub fn pending_len(&self) -> usize {
+    self.pending_messages.0.lock().len()
+  }
+
+  pub fn had_message(&self, is_match: impl Fn(&LspMessage) -> bool) -> bool {
+    self.read_messages.iter().any(&is_match)
+      || self.pending_messages.0.lock().iter().any(&is_match)
+  }
+
+  pub fn read_message<R>(
+    &mut self,
+    mut get_match: impl FnMut(&LspMessage) -> Option<R>,
+  ) -> R {
+    let (msg_queue, cvar) = &*self.pending_messages;
+    let mut msg_queue = msg_queue.lock();
+    loop {
+      for i in 0..msg_queue.len() {
+        let msg = &msg_queue[i];
+        if let Some(result) = get_match(msg) {
+          let msg = msg_queue.remove(i);
+          self.read_messages.push(msg);
+          return result;
+        }
+      }
+      cvar.wait(&mut msg_queue);
+    }
+  }
 }
 
 pub struct LspClient {
   child: Child,
-  reader: io::BufReader<ChildStdout>,
-  /// Used to hold pending messages that have come out of the expected sequence
-  /// by the harness user which will be sent first when trying to consume a
-  /// message before attempting to read a new message.
-  msg_queue: VecDeque<LspMessage>,
+  reader: LspStdoutReader,
   request_id: u64,
   start: Instant,
   writer: io::BufWriter<ChildStdin>,
@@ -179,16 +237,15 @@ impl LspClient {
       command.stderr(Stdio::null());
     }
     let mut child = command.spawn()?;
-
     let stdout = child.stdout.take().unwrap();
-    let reader = io::BufReader::new(stdout);
+    let buf_reader = io::BufReader::new(stdout);
+    let reader = LspStdoutReader::new(buf_reader);
 
     let stdin = child.stdin.take().unwrap();
     let writer = io::BufWriter::new(stdin);
 
     Ok(Self {
       child,
-      msg_queue: VecDeque::new(),
       reader,
       request_id: 1,
       start: Instant::now(),
@@ -202,75 +259,47 @@ impl LspClient {
   }
 
   pub fn queue_is_empty(&self) -> bool {
-    self.msg_queue.is_empty()
+    self.reader.pending_len() == 0
   }
 
   pub fn queue_len(&self) -> usize {
-    self.msg_queue.len()
+    self.reader.pending_len()
   }
 
-  fn read(&mut self) -> Result<LspMessage> {
-    let msg_buf = read_message(&mut self.reader)?;
-    let msg = LspMessage::from(msg_buf.as_slice());
-    Ok(msg)
+  // it's flaky to assert for a notification because a notification
+  // might arrive a little later, so only provide a method for asserting
+  // that there is no notification
+  pub fn assert_no_notification(&mut self, searching_method: &str) {
+    assert!(!self.reader.had_message(|message| match message {
+      LspMessage::Notification(method, _) => method == searching_method,
+      _ => false,
+    }))
   }
 
   pub fn read_notification<R>(&mut self) -> Result<(String, Option<R>)>
   where
     R: de::DeserializeOwned,
   {
-    if !self.msg_queue.is_empty() {
-      let mut msg_queue = VecDeque::new();
-      loop {
-        match self.msg_queue.pop_front() {
-          Some(LspMessage::Notification(method, maybe_params)) => {
-            return notification_result(method, maybe_params)
-          }
-          Some(msg) => msg_queue.push_back(msg),
-          _ => break,
-        }
-      }
-      self.msg_queue = msg_queue;
-    }
-
-    loop {
-      match self.read() {
-        Ok(LspMessage::Notification(method, maybe_params)) => {
-          return notification_result(method, maybe_params)
-        }
-        Ok(msg) => self.msg_queue.push_back(msg),
-        Err(err) => return Err(err),
-      }
-    }
+    self.reader.read_message(|msg| match msg {
+      LspMessage::Notification(method, maybe_params) => Some(
+        notification_result(method.to_owned(), maybe_params.to_owned()),
+      ),
+      _ => None,
+    })
   }
 
   pub fn read_request<R>(&mut self) -> Result<(u64, String, Option<R>)>
   where
     R: de::DeserializeOwned,
   {
-    if !self.msg_queue.is_empty() {
-      let mut msg_queue = VecDeque::new();
-      loop {
-        match self.msg_queue.pop_front() {
-          Some(LspMessage::Request(id, method, maybe_params)) => {
-            return request_result(id, method, maybe_params)
-          }
-          Some(msg) => msg_queue.push_back(msg),
-          _ => break,
-        }
-      }
-      self.msg_queue = msg_queue;
-    }
-
-    loop {
-      match self.read() {
-        Ok(LspMessage::Request(id, method, maybe_params)) => {
-          return request_result(id, method, maybe_params)
-        }
-        Ok(msg) => self.msg_queue.push_back(msg),
-        Err(err) => return Err(err),
-      }
-    }
+    self.reader.read_message(|msg| match msg {
+      LspMessage::Request(id, method, maybe_params) => Some(request_result(
+        *id,
+        method.to_owned(),
+        maybe_params.to_owned(),
+      )),
+      _ => None,
+    })
   }
 
   fn write(&mut self, value: Value) -> Result<()> {
@@ -303,17 +332,17 @@ impl LspClient {
     });
     self.write(value)?;
 
-    loop {
-      match self.read() {
-        Ok(LspMessage::Response(id, maybe_result, maybe_error)) => {
-          assert_eq!(id, self.request_id);
-          self.request_id += 1;
-          return response_result(maybe_result, maybe_error);
-        }
-        Ok(msg) => self.msg_queue.push_back(msg),
-        Err(err) => return Err(err),
+    self.reader.read_message(|msg| match msg {
+      LspMessage::Response(id, maybe_result, maybe_error) => {
+        assert_eq!(*id, self.request_id);
+        self.request_id += 1;
+        Some(response_result(
+          maybe_result.to_owned(),
+          maybe_error.to_owned(),
+        ))
       }
-    }
+      _ => None,
+    })
   }
 
   pub fn write_response<V>(&mut self, id: u64, result: V) -> Result<()>
@@ -351,11 +380,11 @@ mod tests {
   fn test_read_message() {
     let msg1 = b"content-length: 11\r\n\r\nhello world";
     let mut reader1 = std::io::Cursor::new(msg1);
-    assert_eq!(read_message(&mut reader1).unwrap(), b"hello world");
+    assert_eq!(read_message(&mut reader1).unwrap().unwrap(), b"hello world");
 
     let msg2 = b"content-length: 5\r\n\r\nhello world";
     let mut reader2 = std::io::Cursor::new(msg2);
-    assert_eq!(read_message(&mut reader2).unwrap(), b"hello");
+    assert_eq!(read_message(&mut reader2).unwrap().unwrap(), b"hello");
   }
 
   #[test]