diff --git a/crates/ide-lsp/src/caps.rs b/crates/ide-lsp/src/caps.rs index 550736c6..ad84f59e 100644 --- a/crates/ide-lsp/src/caps.rs +++ b/crates/ide-lsp/src/caps.rs @@ -48,7 +48,7 @@ pub fn client_capabilities() -> Value { "semanticTokens": { "refreshSupport": true }, "codeLens": { "refreshSupport": true }, "inlayHint": { "refreshSupport": true }, - "diagnostics": { "refreshSupport": false }, + "diagnostics": { "refreshSupport": true }, "textDocumentContent": { "dynamicRegistration": true } }, "textDocument": { @@ -130,6 +130,7 @@ pub fn client_capabilities() -> Value { "prepareSupportDefaultBehavior": 1, "honorsChangeAnnotations": false }, + "diagnostic": { "dynamicRegistration": false, "relatedDocumentSupport": false }, "publishDiagnostics": { "relatedInformation": true, "tagSupport": { "valueSet": [1, 2] }, diff --git a/crates/ide-lsp/src/handler.rs b/crates/ide-lsp/src/handler.rs index 55130d68..c288623f 100644 --- a/crates/ide-lsp/src/handler.rs +++ b/crates/ide-lsp/src/handler.rs @@ -130,15 +130,8 @@ impl ClientHandler { { return; } - let conv = mgr.conv(self.encoding()); - let items = params - .diagnostics - .iter() - .cloned() - .map(|d| convert::diagnostic(&conv, &path, d, &self.server.name)) - .collect(); - self.server.set_diagnostics(path.clone(), params.diagnostics); - mgr.sink().emit(IdeEvent::Diagnostics { path: path.into(), source: self.server.name.clone(), items }); + let shown = self.server.set_diagnostics(path.clone(), params.diagnostics); + mgr.emit_diagnostics(&self.server, path, shown, self.encoding()); } fn show_message(&self, mgr: &LspManager, params: &Value) { @@ -309,7 +302,11 @@ impl Handler for ClientHandler { mgr.sink().emit(IdeEvent::LspRefresh { server: self.server.id, kind }); Ok(Value::Null) } - "workspace/diagnostic/refresh" | "workspace/inlineValue/refresh" => Ok(Value::Null), + "workspace/diagnostic/refresh" => { + mgr.refresh_pulled_diagnostics(&self.server); + Ok(Value::Null) + } + "workspace/inlineValue/refresh" => Ok(Value::Null), other => Err((codes::METHOD_NOT_FOUND, format!("unsupported method {other}"))), } } diff --git a/crates/ide-lsp/src/inspect.rs b/crates/ide-lsp/src/inspect.rs index 19c4e580..e8509425 100644 --- a/crates/ide-lsp/src/inspect.rs +++ b/crates/ide-lsp/src/inspect.rs @@ -6,8 +6,6 @@ //! (`textDocument/diagnostic`). The others are waited on: the first //! `publishDiagnostics` for the file after it was opened, then a short quiet //! period so follow-up publishes (syntax first, semantics later) are caught. -//! The client never advertises pull support, so servers keep pushing for the -//! editor as before. use crate::convert; use crate::server::Server; diff --git a/crates/ide-lsp/src/lib.rs b/crates/ide-lsp/src/lib.rs index 40665fc3..c18d4814 100644 --- a/crates/ide-lsp/src/lib.rs +++ b/crates/ide-lsp/src/lib.rs @@ -18,6 +18,7 @@ mod inspect; pub mod jdtls; mod launch; mod library_source; +mod pull; mod scope; mod server; mod tcp; @@ -432,6 +433,7 @@ impl LspManager { let wanted: Vec<(DocId, Vec>)> = open.into_iter().map(|(id, language, path)| (id, self.servers_for(&language, &path))).collect(); let mut start = Vec::new(); + let mut opened = Vec::new(); { let mut docs = self.docs.lock(); for (id, servers) in wanted { @@ -446,6 +448,7 @@ impl LspManager { user_did_open(server, &run, &d.path, &d.uri, d.did_open_params()); } attached.opened_gen = Some(run.generation); + opened.push((id, server.clone())); } d.servers.push(attached); if !start.iter().any(|s: &Arc| s.id == server.id) { @@ -456,6 +459,9 @@ impl LspManager { d.servers.sort_by_key(|a| servers.iter().position(|s| s.id == a.server.id).unwrap_or(usize::MAX)); } } + for (doc, server) in opened { + self.schedule_pull(doc, Some(&server), Duration::ZERO); + } for server in &start { self.ensure_started(server); } @@ -611,6 +617,7 @@ impl LspManager { self.refresh_external_project(&server, &run).await; // Publish readiness and (re)open attached documents atomically w.r.t. did_change. + let mut reopened = Vec::new(); { let mut docs = self.docs.lock(); if !server.mark_ready(generation, run.clone()) { @@ -620,16 +627,20 @@ impl LspManager { return; } let open_close = run.caps.read().open_close; - for d in docs.values_mut() { + for (id, d) in docs.iter_mut() { let params = d.did_open_params(); for a in d.servers.iter_mut().filter(|a| a.server.id == server.id && a.opened_gen != Some(generation)) { if open_close && let Err(e) = peer.notify("textDocument/didOpen", params.clone()) { tracing::debug!("{}: didOpen: {e}", server.name); } a.opened_gen = Some(generation); + reopened.push(*id); } } } + for doc in reopened { + self.schedule_pull(doc, Some(&server), Duration::ZERO); + } self.check_clang_tidy(&server).await; server.refresh_busy_status(&self.sink); server.announce_ready(generation); @@ -853,6 +864,7 @@ impl LspManager { } docs.insert(doc, state); } + self.schedule_pull(doc, None, Duration::ZERO); for s in &servers { self.ensure_started(s); if s.running().is_some() @@ -930,6 +942,8 @@ impl LspManager { tracing::debug!("{}: didChange: {e}", server.name); } } + drop(docs); + self.schedule_pull(doc, None, pull::AFTER_EDIT); } pub fn did_save(&self, doc: DocId, text: &str) { @@ -952,11 +966,15 @@ impl LspManager { tracing::debug!("{}: didSave: {e}", a.server.name); } } + drop(docs); + self.schedule_pull(doc, None, Duration::ZERO); + self.pull_dependents(doc); } pub fn did_close(&self, doc: DocId) { let Some(d) = self.docs.lock().remove(&doc) else { return }; for a in &d.servers { + self.forget_pulled(&a.server, &d.path, a.server.running().as_ref()); let Some(run) = a.server.running() else { continue }; if a.opened_gen == Some(run.generation) && run.caps.read().open_close diff --git a/crates/ide-lsp/src/pull.rs b/crates/ide-lsp/src/pull.rs new file mode 100644 index 00000000..63016c0e --- /dev/null +++ b/crates/ide-lsp/src/pull.rs @@ -0,0 +1,192 @@ +//! Pull diagnostics (`textDocument/diagnostic`). Servers that provide them may stop pushing +//! their own analysis to a client that can pull (rust-analyzer pushes only `cargo check` +//! then): the editor pulls an open file's diagnostics when it opens, after edits and saves, +//! and when the server asks for a refresh. Pulled diagnostics are kept apart from pushed ones +//! and shown together. + +use crate::server::{Running, Server}; +use crate::{LspManager, convert}; +use ide_core::{DocId, IdeEvent}; +use ide_jsonrpc::RpcError; +use lsp_types as lsp; +use serde_json::{Value, json}; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +/// How long typing must pause before a changed file is pulled again. +pub(crate) const AFTER_EDIT: Duration = Duration::from_millis(300); +const TIMEOUT: Duration = Duration::from_secs(30); +/// Retries of a pull the server cancelled (`ServerCancelled`, e.g. while indexing). +const RETRIES: u32 = 20; + +struct PullTarget { + server: Arc, + run: Running, + path: PathBuf, + uri: lsp::Uri, +} + +fn provides_pull(run: &Running) -> bool { + let caps = run.caps.read(); + caps.restriction.diagnostics() && caps.has("diagnosticProvider") +} + +fn inter_file(run: &Running) -> bool { + run.caps.read().raw.pointer("/diagnosticProvider/interFileDependencies").and_then(Value::as_bool) == Some(true) +} + +impl LspManager { + /// Pulls `doc`'s diagnostics from its servers that provide them (`only`: from that one) + /// after `delay`; a later pull of the same file supersedes it. + pub(crate) fn schedule_pull(&self, doc: DocId, only: Option<&Server>, delay: Duration) { + let targets: Vec = { + let docs = self.docs.lock(); + let Some(d) = docs.get(&doc) else { return }; + d.servers + .iter() + .filter(|a| only.is_none_or(|s| s.id == a.server.id)) + .filter_map(|a| { + let run = a.server.running()?; + (a.opened_gen == Some(run.generation) && provides_pull(&run)).then(|| PullTarget { + server: a.server.clone(), + run, + path: d.path.clone(), + uri: d.uri.clone(), + }) + }) + .collect() + }; + for t in targets { + if t.server.suppresses_diagnostics(&t.path) { + continue; + } + let pull = t.server.next_pull(&t.path); + let me = self.me.clone(); + self.runtime.spawn(async move { + tokio::time::sleep(delay).await; + if let Some(me) = me.upgrade() { + me.pull(t, pull).await; + } + }); + } + } + + /// Pulls every open file of `server` again (`workspace/diagnostic/refresh`). + pub(crate) fn refresh_pulled_diagnostics(&self, server: &Server) { + let docs: Vec = self + .docs + .lock() + .iter() + .filter(|(_, d)| d.servers.iter().any(|a| a.server.id == server.id)) + .map(|(id, _)| *id) + .collect(); + for doc in docs { + self.schedule_pull(doc, Some(server), Duration::ZERO); + } + } + + /// After `doc` was saved: servers whose diagnostics depend on other files pull those + /// open files too. + pub(crate) fn pull_dependents(&self, doc: DocId) { + let docs = self.docs.lock(); + let Some(d) = docs.get(&doc) else { return }; + let servers: Vec> = d + .servers + .iter() + .filter(|a| a.server.running().is_some_and(|run| provides_pull(&run) && inter_file(&run))) + .map(|a| a.server.clone()) + .collect(); + let others: Vec<(DocId, Vec>)> = docs + .iter() + .filter(|(id, _)| **id != doc) + .map(|(id, o)| { + (*id, servers.iter().filter(|s| o.servers.iter().any(|a| a.server.id == s.id)).cloned().collect()) + }) + .collect(); + drop(docs); + for (other, servers) in others { + for server in servers { + self.schedule_pull(other, Some(&server), Duration::ZERO); + } + } + } + + /// Drops what was pulled for a closed file and shows what remains. + pub(crate) fn forget_pulled(&self, server: &Server, path: &std::path::Path, run: Option<&Running>) { + if let Some(shown) = server.clear_pulled(path) { + let enc = run.map(|r| r.caps.read().encoding).unwrap_or_default(); + self.emit_diagnostics(server, path.to_owned(), shown, enc); + } + } + + async fn pull(&self, t: PullTarget, pull: u64) { + let mut attempt = 0; + loop { + if !t.server.is_latest_pull(&t.path, pull) { + return; + } + let mut params = json!({ "textDocument": { "uri": t.uri } }); + if let Some(id) = t.server.pulled_result_id(&t.path) { + params["previousResultId"] = json!(id); + } + if let Some(identifier) = t.run.caps.read().raw.pointer("/diagnosticProvider/identifier") { + params["identifier"] = identifier.clone(); + } + let request = t.run.peer.request::("textDocument/diagnostic", params, None); + let report = match tokio::time::timeout(TIMEOUT, request).await { + Ok(Ok(report)) => report, + // This request has no cancellation of its own: the server cancelled it + // (`ServerCancelled` while busy, which asks to be asked again). + Ok(Err(RpcError::Cancelled)) if attempt < RETRIES => { + attempt += 1; + tokio::time::sleep(Duration::from_millis(250 * u64::from(attempt.min(8)))).await; + continue; + } + Ok(Err(e)) => { + tracing::debug!("{}: textDocument/diagnostic: {e}", t.server.name); + return; + } + Err(_) => { + tracing::debug!("{}: textDocument/diagnostic timed out", t.server.name); + return; + } + }; + // A newer pull, a closed file or a restarted server makes this report stale. + if !t.server.is_latest_pull(&t.path, pull) + || t.server.running().is_none_or(|r| r.generation != t.run.generation) + || !self.docs.lock().values().any(|d| d.path == t.path) + { + return; + } + let result_id = report.get("resultId").and_then(Value::as_str).map(str::to_owned); + let items = match report.get("kind").and_then(Value::as_str) { + Some("unchanged") => None, + _ => match serde_json::from_value(report.get("items").cloned().unwrap_or(json!([]))) { + Ok(items) => Some(items), + Err(e) => { + tracing::debug!("{}: bad textDocument/diagnostic report: {e}", t.server.name); + return; + } + }, + }; + let shown = t.server.set_pulled(t.path.clone(), result_id, items); + let enc = t.run.caps.read().encoding; + self.emit_diagnostics(&t.server, t.path, shown, enc); + return; + } + } + + /// Shows `items` as `server`'s diagnostics for `path`. + pub(crate) fn emit_diagnostics( + &self, + server: &Server, + path: PathBuf, + items: Vec, + enc: crate::text::Encoding, + ) { + let conv = self.conv(enc); + let items = items.into_iter().map(|d| convert::diagnostic(&conv, &path, d, &server.name)).collect(); + self.sink.emit(IdeEvent::Diagnostics { path: path.into(), source: server.name.clone(), items }); + } +} diff --git a/crates/ide-lsp/src/server.rs b/crates/ide-lsp/src/server.rs index 76f08cc6..6d57b887 100644 --- a/crates/ide-lsp/src/server.rs +++ b/crates/ide-lsp/src/server.rs @@ -97,7 +97,13 @@ pub struct Server { started: Mutex>, stderr: Mutex>, pub progress: Mutex>, + /// What the editor shows per path: the pushed and the pulled diagnostics together. diagnostics: Mutex>>, + /// `publishDiagnostics` per path. + pushed: Mutex>>, + /// `textDocument/diagnostic` reports per path, and the latest pull asked for each. + pulled: Mutex>, + pulls: Mutex>, /// When diagnostics were last published per path (empty lists included). published: Mutex>, /// Woken on every `publishDiagnostics` (Inspect Code waits on it). @@ -132,6 +138,9 @@ impl Server { stderr: Mutex::new(VecDeque::new()), progress: Mutex::new(HashMap::new()), diagnostics: Mutex::new(HashMap::new()), + pushed: Mutex::new(HashMap::new()), + pulled: Mutex::new(HashMap::new()), + pulls: Mutex::new(HashMap::new()), published: Mutex::new(HashMap::new()), published_notify: Notify::new(), idle_notify: Notify::new(), @@ -493,17 +502,71 @@ impl Server { self.progress.lock().drain().map(|(_, p)| (p.task, p.title)).collect() } - pub fn set_diagnostics(&self, path: PathBuf, items: Vec) { + /// Records the diagnostics the server pushed for `path`; returns what the file now shows. + pub fn set_diagnostics(&self, path: PathBuf, items: Vec) -> Vec { + set_or_remove(&mut self.pushed.lock(), path.clone(), items); + self.combine(path) + } + + /// Records a pulled report for `path` (`None` items: unchanged since `result_id`'s + /// predecessor); returns what the file now shows. + pub fn set_pulled( + &self, + path: PathBuf, + result_id: Option, + items: Option>, + ) -> Vec { + { + let mut pulled = self.pulled.lock(); + let entry = pulled.entry(path.clone()).or_default(); + entry.result_id = result_id; + if let Some(items) = items { + entry.items = items; + } + } + self.combine(path) + } + + /// Forgets the pulled report of a closed file; `None` when there was none. + pub fn clear_pulled(&self, path: &Path) -> Option> { + self.pulls.lock().remove(path); + self.pulled.lock().remove(path)?; + Some(self.combine(path.to_owned())) + } + + /// The `resultId` of the last report pulled for `path`. + pub fn pulled_result_id(&self, path: &Path) -> Option { + self.pulled.lock().get(path).and_then(|p| p.result_id.clone()) + } + + /// Starts a pull for `path`, superseding earlier ones; see [`Server::is_latest_pull`]. + pub fn next_pull(&self, path: &Path) -> u64 { + static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + let id = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.pulls.lock().insert(path.to_owned(), id); + id + } + + pub fn is_latest_pull(&self, path: &Path, id: u64) -> bool { + self.pulls.lock().get(path) == Some(&id) + } + + fn combine(&self, path: PathBuf) -> Vec { + let mut items = self.pushed.lock().get(&path).cloned().unwrap_or_default(); + if let Some(pulled) = self.pulled.lock().get(&path) { + for d in &pulled.items { + if !items.contains(d) { + items.push(d.clone()); + } + } + } { let mut d = self.diagnostics.lock(); self.published.lock().insert(path.clone(), Instant::now()); - if items.is_empty() { - d.remove(&path); - } else { - d.insert(path, items); - } + set_or_remove(&mut d, path, items.clone()); } self.published_notify.notify_waiters(); + items } /// When diagnostics for `path` were last published by this server. @@ -528,10 +591,28 @@ impl Server { pub fn take_diagnostic_paths(&self) -> Vec { self.published.lock().clear(); + self.pushed.lock().clear(); + self.pulled.lock().clear(); + self.pulls.lock().clear(); self.diagnostics.lock().drain().map(|(p, _)| p).collect() } } +/// A file's last `textDocument/diagnostic` report. +#[derive(Default)] +struct Pulled { + result_id: Option, + items: Vec, +} + +fn set_or_remove(map: &mut HashMap>, path: PathBuf, items: Vec) { + if items.is_empty() { + map.remove(&path); + } else { + map.insert(path, items); + } +} + /// The last `n` stderr lines, led by the exception headline: Node.js / Python / /// JVM crashes print the error first and a stack after it, so a plain tail /// shows frames without saying what went wrong. diff --git a/crates/ide-lsp/tests/mock_server.rs b/crates/ide-lsp/tests/mock_server.rs index cf7d0fd8..deca000f 100644 --- a/crates/ide-lsp/tests/mock_server.rs +++ b/crates/ide-lsp/tests/mock_server.rs @@ -768,6 +768,113 @@ async fn diagnostics_become_events_and_feed_code_actions() { assert_eq!(err.kind, ide_core::error::ErrorKind::Unavailable); } +/// The diagnostics shown last for `path` by `source`, as their messages. +fn shown(h: &Harness, path: &Path, source: &str) -> Option> { + h.events.lock().iter().rev().find_map(|e| match e { + IdeEvent::Diagnostics { path: p, source: s, items } if **p == *path && s == source => { + Some(items.iter().map(|d| d.message.clone()).collect()) + } + _ => None, + }) +} + +/// Reports a "bad" diagnostic per line containing `bad`, as `textDocument/diagnostic`; answers +/// `unchanged` when the text is what `previousResultId` was computed from. +fn pull_responder() -> Responder { + Arc::new(|s: &MockState, method: &str, params: &Value| { + if method != "textDocument/diagnostic" { + return None; + } + let text = s.docs.lock().get(params["textDocument"]["uri"].as_str()?).cloned()?; + let id = format!("{}", text.len() * 31 + text.matches("bad").count()); + if params["previousResultId"].as_str() == Some(id.as_str()) { + return Some(json!({ "kind": "unchanged", "resultId": id })); + } + let items: Vec = text + .lines() + .enumerate() + .filter(|(_, l)| l.contains("bad")) + .map(|(i, _)| { + json!({ "range": {"start":{"line":i,"character":0},"end":{"line":i,"character":3}}, + "severity": 1, "message": "bad" }) + }) + .collect(); + Some(json!({ "kind": "full", "resultId": id, "items": items })) + }) +} + +#[tokio::test(flavor = "multi_thread")] +async fn pulled_diagnostics_follow_edits_and_join_pushed_ones() { + let mut pull_caps = incremental_caps(); + pull_caps["diagnosticProvider"] = json!({ "interFileDependencies": false, "workspaceDiagnostics": false }); + let h = harness(&[("mock", pull_caps), ("push", incremental_caps())]); + let s = h.launcher.state("mock"); + *s.responder.lock() = Some(pull_responder()); + let (doc, path) = h.open("a.mk", "ok\nbad\n"); + h.wait_status("mock", ServerStatus::Ready).await; + h.wait_status("push", ServerStatus::Ready).await; + assert!(s.requested("initialize")[0]["capabilities"]["textDocument"]["diagnostic"].is_object()); + eventually(|| shown(&h, &path, "mock") == Some(vec!["bad".into()])).await; + assert!(h.launcher.state("push").requested("textDocument/diagnostic").is_empty(), "it pulls from providers only"); + + // What the server still pushes (rust-analyzer: `cargo check`) shows next to what was pulled. + let pushed = json!({ "range": {"start":{"line":0,"character":0},"end":{"line":0,"character":2}}, + "severity": 2, "message": "pushed" }); + s.peer() + .notify("textDocument/publishDiagnostics", json!({ "uri": uri_of(&path), "diagnostics": [pushed] })) + .unwrap(); + eventually(|| shown(&h, &path, "mock") == Some(vec!["pushed".into(), "bad".into()])).await; + + // Edits are pulled again once typing pauses. + h.mgr.did_change(doc, 2, &[change(rng(0, 0, 0, 0), "bad\n")], &|| unreachable!()); + eventually(|| shown(&h, &path, "mock") == Some(vec!["pushed".into(), "bad".into(), "bad".into()])).await; + // Pulled diagnostics are code actions' context like pushed ones. + let _ = h.mgr.code_actions(doc, rng(2, 1, 2, 1), vec![]).await; + let ctx = &s.requested("textDocument/codeAction")[0]["context"]["diagnostics"]; + assert_eq!(ctx.as_array().unwrap().iter().map(|d| d["message"].as_str().unwrap()).collect::>(), ["bad"]); + + // Saving unchanged text asks with the last result id; `unchanged` keeps what is shown. + let pulls = s.requested("textDocument/diagnostic").len(); + h.mgr.did_save(doc, "bad\nok\nbad\n"); + eventually(|| s.requested("textDocument/diagnostic").len() > pulls).await; + let last = s.requested("textDocument/diagnostic").last().cloned().unwrap(); + assert!(last["previousResultId"].is_string()); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(shown(&h, &path, "mock"), Some(vec!["pushed".into(), "bad".into(), "bad".into()])); + + // Closing drops what was pulled; the pushed diagnostics stay the server's to clear. + h.mgr.did_close(doc); + eventually(|| shown(&h, &path, "mock") == Some(vec!["pushed".into()])).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_cancelled_pull_is_retried_and_a_refresh_pulls_again() { + let mut caps = incremental_caps(); + caps["diagnosticProvider"] = json!({ "interFileDependencies": false, "workspaceDiagnostics": false }); + let h = harness(&[("mock", caps)]); + let s = h.launcher.state("mock"); + *s.responder.lock() = Some(pull_responder()); + // Busy indexing: the server cancels and asks to be asked again. + s.errors.lock().insert("textDocument/diagnostic".into(), (-32802, "indexing".into())); + let (_doc, path) = h.open("a.mk", "bad\n"); + h.wait_status("mock", ServerStatus::Ready).await; + eventually(|| s.requested("textDocument/diagnostic").len() >= 2).await; + assert_eq!(shown(&h, &path, "mock"), None); + s.errors.lock().clear(); + eventually(|| shown(&h, &path, "mock") == Some(vec!["bad".into()])).await; + + // `workspace/diagnostic/refresh` (e.g. after a project reload) pulls open files again. + *s.responder.lock() = Some(Arc::new(|_: &MockState, method: &str, _: &Value| { + (method == "textDocument/diagnostic").then(|| { + json!({ "kind": "full", "items": [ + { "range": {"start":{"line":0,"character":0},"end":{"line":0,"character":3}}, "message": "reloaded" } ] }) + }) + })); + let refreshed: Value = s.peer().request("workspace/diagnostic/refresh", Value::Null, None).await.unwrap(); + assert_eq!(refreshed, Value::Null); + eventually(|| shown(&h, &path, "mock") == Some(vec!["reloaded".into()])).await; +} + #[tokio::test(flavor = "multi_thread")] async fn apply_edit_hook_and_server_requests() { let h = harness(&[("mock", incremental_caps())]);