From 8f7ab3d3453ab615601f9827fd009385b88e6289 Mon Sep 17 00:00:00 2001 From: "Henning P. Schmiedehausen" Date: Mon, 28 Sep 2026 22:14:10 -0700 Subject: [PATCH 1/2] fix(diagnostics): refresh dependent documents after file changes --- .../src/context/diagnostic_service.rs | 74 +++++++++ .../emmylua_ls/src/handlers/diagnostic/mod.rs | 2 +- .../text_document/text_document_handler.rs | 12 ++ .../text_document/watched_file_handler.rs | 26 +++- crates/emmylua_ls/tests/diagnostic_refresh.py | 140 ++++++++++++++++++ 5 files changed, 247 insertions(+), 7 deletions(-) create mode 100644 crates/emmylua_ls/tests/diagnostic_refresh.py diff --git a/crates/emmylua_ls/src/context/diagnostic_service.rs b/crates/emmylua_ls/src/context/diagnostic_service.rs index 6369281a7..586d3f8d7 100644 --- a/crates/emmylua_ls/src/context/diagnostic_service.rs +++ b/crates/emmylua_ls/src/context/diagnostic_service.rs @@ -18,6 +18,7 @@ pub struct DiagnosticService { client: Arc, status_bar: Arc, diagnostic_tokens: Arc>>, + refresh_token: Mutex>, workspace_diagnostic_token: Arc>>, workspace_run_lock: Arc>, workspace_diagnostic_done: AtomicBool, @@ -34,6 +35,7 @@ impl DiagnosticService { analysis, client, diagnostic_tokens: Arc::new(Mutex::new(HashMap::new())), + refresh_token: Mutex::new(None), workspace_diagnostic_token: Arc::new(Mutex::new(None)), workspace_run_lock: Arc::new(Mutex::new(())), workspace_diagnostic_done: AtomicBool::new(false), @@ -42,6 +44,26 @@ impl DiagnosticService { } } + /// Coalesce cross-file invalidations after analysis has incorporated an edit. + pub async fn schedule_diagnostic_refresh(&self, interval: u64) { + let mut pending = self.refresh_token.lock().await; + if let Some(token) = pending.take() { + token.cancel(); + } + let token = CancellationToken::new(); + *pending = Some(token.clone()); + let client = self.client.clone(); + tokio::spawn(async move { + tokio::select! { + biased; + _ = token.cancelled() => {}, + _ = tokio::time::sleep(Duration::from_millis(interval)) => { + client.refresh_workspace_diagnostics(); + } + } + }); + } + pub async fn add_diagnostic_task(&self, file_id: FileId, interval: u64) { let mut tokens = self.diagnostic_tokens.lock().await; @@ -385,3 +407,55 @@ async fn run_workspace_batch( (result, count == valid_file_count) } + +impl Drop for DiagnosticService { + fn drop(&mut self) { + if let Some(token) = self.refresh_token.get_mut().take() { + token.cancel(); + } + } +} + +#[cfg(test)] +mod refresh_tests { + use super::*; + use lsp_server::{Connection, Message}; + + fn diagnostic() -> (DiagnosticService, Connection) { + let (server, client_connection) = Connection::memory(); + let client = Arc::new(ClientProxy::new(Arc::new(server))); + let status = Arc::new(StatusBar::new(client.clone(), false)); + ( + DiagnosticService::new(Arc::new(AnalysisState::new()), status, client), + client_connection, + ) + } + + #[tokio::test] + async fn refresh_coalesces_and_is_scoped_to_its_client() { + let (first, client) = diagnostic(); + let (_second, other_client) = diagnostic(); + for _ in 0..10 { + first.schedule_diagnostic_refresh(30).await; + } + tokio::time::sleep(Duration::from_millis(100)).await; + let Message::Request(request) = client.receiver.try_recv().unwrap() else { + panic!("expected request") + }; + assert_eq!(request.method, "workspace/diagnostic/refresh"); + assert!(client.receiver.try_recv().is_err()); + assert!(other_client.receiver.try_recv().is_err()); + first.schedule_diagnostic_refresh(0).await; + tokio::time::sleep(Duration::from_millis(30)).await; + assert!(client.receiver.try_recv().is_ok()); + } + + #[tokio::test] + async fn dropping_diagnostics_cancels_pending_refresh() { + let (diagnostic, client) = diagnostic(); + diagnostic.schedule_diagnostic_refresh(30).await; + drop(diagnostic); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(client.receiver.try_recv().is_err()); + } +} diff --git a/crates/emmylua_ls/src/handlers/diagnostic/mod.rs b/crates/emmylua_ls/src/handlers/diagnostic/mod.rs index 04eeff257..c88835d26 100644 --- a/crates/emmylua_ls/src/handlers/diagnostic/mod.rs +++ b/crates/emmylua_ls/src/handlers/diagnostic/mod.rs @@ -15,7 +15,7 @@ impl RegisterCapabilities for DiagnosticCapabilities { server_capabilities.diagnostic_provider = Some(DiagnosticServerCapabilities::Options(DiagnosticOptions { identifier: Some("EmmyLua".to_string()), - inter_file_dependencies: false, + inter_file_dependencies: true, workspace_diagnostics: true, ..Default::default() })) diff --git a/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs b/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs index 714f00951..2f1516197 100644 --- a/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs +++ b/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs @@ -64,6 +64,12 @@ pub async fn process_did_open_text_document( let supports_pull = context.lsp_features().supports_pull_diagnostic(); // Schedule diagnostic task without holding any locks + if supports_pull && context.lsp_features().supports_refresh_diagnostic() && file_id.is_some() { + context + .file_diagnostic() + .schedule_diagnostic_refresh(interval) + .await; + } if !supports_pull { if let Some(file_id) = file_id { context @@ -166,6 +172,12 @@ pub async fn process_did_change_text_document( let supports_pull = context.lsp_features().supports_pull_diagnostic(); // Schedule diagnostic task + if supports_pull && context.lsp_features().supports_refresh_diagnostic() && file_id.is_some() { + context + .file_diagnostic() + .schedule_diagnostic_refresh(interval) + .await; + } if !supports_pull { if let Some(file_id) = file_id { context diff --git a/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs b/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs index 0898ac456..a3a7c0e65 100644 --- a/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs +++ b/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs @@ -86,19 +86,33 @@ pub async fn process_did_change_watched_files( .await; } - let file_ids = context + let (file_ids, removed_file) = context .analysis() .update(|analysis| { + let mut removed_file = false; for uri in &removed_uris { + removed_file |= analysis.get_file_id(uri).is_some(); analysis.remove_file_by_uri(uri); } - analysis.update_files_by_uri(watched_lua_files) + ( + analysis.update_files_by_uri(watched_lua_files), + removed_file, + ) }) .await; - context - .file_diagnostic() - .add_files_diagnostic_task(file_ids, interval) - .await; + if lsp_features.supports_pull_diagnostic() { + if (removed_file || !file_ids.is_empty()) && lsp_features.supports_refresh_diagnostic() { + context + .file_diagnostic() + .schedule_diagnostic_refresh(interval) + .await; + } + } else { + context + .file_diagnostic() + .add_files_diagnostic_task(file_ids, interval) + .await; + } Some(()) } diff --git a/crates/emmylua_ls/tests/diagnostic_refresh.py b/crates/emmylua_ls/tests/diagnostic_refresh.py new file mode 100644 index 000000000..198fc895a --- /dev/null +++ b/crates/emmylua_ls/tests/diagnostic_refresh.py @@ -0,0 +1,140 @@ +"""Protocol regression: python3 crates/emmylua_ls/tests/diagnostic_refresh.py target/release/emmylua_ls. + +Uses only the standard library. Fixtures and server logs stay under build/. +""" +import json +import queue +import subprocess +import sys +import tempfile +import threading +import time +from pathlib import Path + + +class Client: + def __init__(self, binary, root, refresh): + self.root = root + self.messages = queue.Queue() + self.next_id = 0 + self.log = (root / 'server.log').open('w') + self.process = subprocess.Popen([str(binary)], cwd=root, stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=self.log) + threading.Thread(target=self.read, daemon=True).start() + result = self.request('initialize', { + 'processId': None, 'rootUri': root.as_uri(), + 'workspaceFolders': [{'uri': root.as_uri(), 'name': 'test'}], + 'capabilities': {'textDocument': {'diagnostic': {}}, + 'workspace': {'diagnostics': {'refreshSupport': refresh}}}}) + assert result['capabilities']['diagnosticProvider']['interFileDependencies'] + self.send('initialized', {}) + time.sleep(1) + + def read(self): + while True: + headers = {} + while True: + line = self.process.stdout.readline() + if not line: + return + if line == b'\r\n': + break + key, value = line.decode().split(':', 1) + headers[key.lower()] = value.strip() + self.messages.put(json.loads(self.process.stdout.read(int(headers['content-length'])))) + + def write(self, message): + raw = json.dumps({'jsonrpc': '2.0', **message}).encode() + self.process.stdin.write(f'Content-Length: {len(raw)}\r\n\r\n'.encode() + raw) + self.process.stdin.flush() + + def send(self, method, params): + self.write({'method': method, 'params': params}) + + def receive(self, timeout): + message = self.messages.get(timeout=timeout) + if 'method' in message and 'id' in message: + result = [] if message['method'] == 'workspace/configuration' else None + self.write({'id': message['id'], 'result': result}) + return message + + def request(self, method, params): + self.next_id += 1 + self.write({'id': self.next_id, 'method': method, 'params': params}) + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + message = self.receive(max(0.01, deadline - time.monotonic())) + if message.get('id') == self.next_id and 'method' not in message: + assert 'error' not in message, message + return message['result'] + raise AssertionError('Response timed out') + + def refreshes(self, duration=0.6): + count = 0 + deadline = time.monotonic() + duration + while time.monotonic() < deadline: + try: + message = self.receive(max(0.01, deadline - time.monotonic())) + except queue.Empty: + break + count += message.get('method') == 'workspace/diagnostic/refresh' + return count + + def diagnostic(self): + return self.request('textDocument/diagnostic', { + 'textDocument': {'uri': (self.root / 'settings.lua').as_uri()}})['items'] + + def close(self): + self.process.terminate() + self.process.wait(timeout=5) + self.process.stdin.close() + self.process.stdout.close() + self.log.close() + + +def run(binary, root, refresh): + required = '---@class Provider\n---@field set_values fun()\n' + optional = '---@class Provider\n-- ---@field set_values fun()\n' + consumer = '---@type Provider\nlocal settings_table = {}\nreturn settings_table\n' + (root / 'types.lua').write_text(required) + (root / 'settings.lua').write_text(consumer) + (root / '.emmyrc.json').write_text('{"diagnostics":{"diagnosticInterval":100}}') + client = Client(binary, root, refresh) + try: + for name, text in [('types.lua', required), ('settings.lua', consumer)]: + client.send('textDocument/didOpen', {'textDocument': { + 'uri': (root / name).as_uri(), 'languageId': 'lua', 'version': 1, 'text': text}}) + client.refreshes() + assert any(d.get('code') == 'missing-fields' for d in client.diagnostic()) + version = 1 + for text, missing in [(optional, False), (required, True)]: + for _ in range(10): + version += 1 + client.send('textDocument/didChange', {'textDocument': { + 'uri': (root / 'types.lua').as_uri(), 'version': version}, + 'contentChanges': [{'text': text}]}) + assert client.refreshes() == int(refresh), 'Edits must coalesce and respect capabilities' + assert any(d.get('code') == 'missing-fields' for d in client.diagnostic()) == missing + client.send('textDocument/didClose', {'textDocument': {'uri': (root / 'types.lua').as_uri()}}) + for kind, text, missing in [(2, optional, False), (2, required, True), + (3, None, False), (1, required, True)]: + path = root / 'types.lua' + if text is None: + path.unlink() + else: + path.write_text(text) + client.send('workspace/didChangeWatchedFiles', { + 'changes': [{'uri': path.as_uri(), 'type': kind}]}) + assert client.refreshes() == int(refresh), 'Watcher refresh must respect capabilities' + assert any(d.get('code') == 'missing-fields' for d in client.diagnostic()) == missing + finally: + client.close() + + +if __name__ == '__main__': + binary = Path(sys.argv[1]).resolve() + Path('build').mkdir(exist_ok=True) + for refresh in (True, False): + with tempfile.TemporaryDirectory(prefix='diagnostic-refresh-', dir=Path('build').resolve()) as root: + run(binary, Path(root), refresh) + print('PASS: remove/restore required field, debounce, watcher change/delete/create, capability gating') From 9e8ed55d99c51d4239ff4b4fd380c0f0dffba8a1 Mon Sep 17 00:00:00 2001 From: "Henning P. Schmiedehausen" Date: Mon, 12 Oct 2026 10:58:40 -0700 Subject: [PATCH 2/2] fix(diagnostics): harden refresh lifecycle and coverage Cancel pending refreshes before acknowledging shutdown, reject new work from retained contexts, and avoid keeping the client alive in timer tasks. Serialize final sends with shutdown and document zero-delay behavior. Refresh after tracked-file removal on close and file renames. Route configuration reloads through the capability-gated refresh lifecycle. Use a paused clock for debounce tests and synchronize protocol tests on analysis state. Cover close, rename, configuration reload, capability combinations and shutdown, with explicit handling of client requests. --- crates/emmylua_ls/Cargo.toml | 1 + .../src/context/diagnostic_service.rs | 125 +++++++-- crates/emmylua_ls/src/context/mod.rs | 5 +- crates/emmylua_ls/src/context/snapshot.rs | 11 + .../src/context/workspace_manager.rs | 20 +- .../text_document/text_document_handler.rs | 66 ++--- .../text_document/watched_file_handler.rs | 7 +- .../handlers/workspace/did_rename_files.rs | 14 +- .../src/server/message_processor.rs | 8 + crates/emmylua_ls/tests/diagnostic_refresh.py | 249 +++++++++++++----- 10 files changed, 359 insertions(+), 147 deletions(-) diff --git a/crates/emmylua_ls/Cargo.toml b/crates/emmylua_ls/Cargo.toml index 335eef2a5..5a1413b3f 100644 --- a/crates/emmylua_ls/Cargo.toml +++ b/crates/emmylua_ls/Cargo.toml @@ -53,6 +53,7 @@ optional = true [dev-dependencies] googletest.workspace = true +tokio = { workspace = true, features = ["test-util"] } [[bin]] diff --git a/crates/emmylua_ls/src/context/diagnostic_service.rs b/crates/emmylua_ls/src/context/diagnostic_service.rs index 586d3f8d7..6e0f84d6b 100644 --- a/crates/emmylua_ls/src/context/diagnostic_service.rs +++ b/crates/emmylua_ls/src/context/diagnostic_service.rs @@ -18,7 +18,8 @@ pub struct DiagnosticService { client: Arc, status_bar: Arc, diagnostic_tokens: Arc>>, - refresh_token: Mutex>, + refresh_token: Arc>>, + refresh_shutdown: CancellationToken, workspace_diagnostic_token: Arc>>, workspace_run_lock: Arc>, workspace_diagnostic_done: AtomicBool, @@ -35,7 +36,8 @@ impl DiagnosticService { analysis, client, diagnostic_tokens: Arc::new(Mutex::new(HashMap::new())), - refresh_token: Mutex::new(None), + refresh_token: Arc::new(Mutex::new(None)), + refresh_shutdown: CancellationToken::new(), workspace_diagnostic_token: Arc::new(Mutex::new(None)), workspace_run_lock: Arc::new(Mutex::new(())), workspace_diagnostic_done: AtomicBool::new(false), @@ -45,25 +47,51 @@ impl DiagnosticService { } /// Coalesce cross-file invalidations after analysis has incorporated an edit. + /// `interval` is in milliseconds; zero requests a refresh without a debounce delay. pub async fn schedule_diagnostic_refresh(&self, interval: u64) { let mut pending = self.refresh_token.lock().await; + if self.refresh_shutdown.is_cancelled() { + return; + } if let Some(token) = pending.take() { token.cancel(); } let token = CancellationToken::new(); *pending = Some(token.clone()); - let client = self.client.clone(); + drop(pending); + let client = Arc::downgrade(&self.client); + let pending = self.refresh_token.clone(); + let shutdown = self.refresh_shutdown.clone(); + let deadline = tokio::time::Instant::now() + Duration::from_millis(interval); tokio::spawn(async move { tokio::select! { biased; + _ = shutdown.cancelled() => {}, _ = token.cancelled() => {}, - _ = tokio::time::sleep(Duration::from_millis(interval)) => { - client.refresh_workspace_diagnostics(); + _ = tokio::time::sleep_until(deadline) => { + // Serialize the final send with replacement and shutdown, so no + // refresh can be sent after shutdown_diagnostic_refresh returns. + let mut pending = pending.lock().await; + if !shutdown.is_cancelled() && !token.is_cancelled() { + if let Some(client) = client.upgrade() { + client.refresh_workspace_diagnostics(); + } + pending.take(); + } } } }); } + /// Stop pending refreshes and reject scheduling by outstanding update tasks. + pub async fn shutdown_diagnostic_refresh(&self) { + let mut pending = self.refresh_token.lock().await; + self.refresh_shutdown.cancel(); + if let Some(token) = pending.take() { + token.cancel(); + } + } + pub async fn add_diagnostic_task(&self, file_id: FileId, interval: u64) { let mut tokens = self.diagnostic_tokens.lock().await; @@ -410,52 +438,103 @@ async fn run_workspace_batch( impl Drop for DiagnosticService { fn drop(&mut self) { - if let Some(token) = self.refresh_token.get_mut().take() { - token.cancel(); - } + self.refresh_shutdown.cancel(); } } #[cfg(test)] mod refresh_tests { use super::*; + use crate::context::ServerContext; use lsp_server::{Connection, Message}; - fn diagnostic() -> (DiagnosticService, Connection) { + fn diagnostic() -> (Arc, Connection) { let (server, client_connection) = Connection::memory(); let client = Arc::new(ClientProxy::new(Arc::new(server))); let status = Arc::new(StatusBar::new(client.clone(), false)); ( - DiagnosticService::new(Arc::new(AnalysisState::new()), status, client), + Arc::new(DiagnosticService::new( + Arc::new(AnalysisState::new()), + status, + client, + )), client_connection, ) } - #[tokio::test] + async fn advance(milliseconds: u64) { + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_millis(milliseconds)).await; + tokio::task::yield_now().await; + } + + fn expect_refresh(client: &Connection) { + let Message::Request(request) = client.receiver.try_recv().unwrap() else { + panic!("expected request") + }; + assert_eq!(request.method, "workspace/diagnostic/refresh"); + assert!(client.receiver.try_recv().is_err()); + } + + #[tokio::test(start_paused = true)] async fn refresh_coalesces_and_is_scoped_to_its_client() { let (first, client) = diagnostic(); let (_second, other_client) = diagnostic(); for _ in 0..10 { - first.schedule_diagnostic_refresh(30).await; + first.schedule_diagnostic_refresh(1000).await; } - tokio::time::sleep(Duration::from_millis(100)).await; - let Message::Request(request) = client.receiver.try_recv().unwrap() else { - panic!("expected request") - }; - assert_eq!(request.method, "workspace/diagnostic/refresh"); + advance(999).await; assert!(client.receiver.try_recv().is_err()); + advance(1).await; + expect_refresh(&client); assert!(other_client.receiver.try_recv().is_err()); + assert!(first.refresh_token.lock().await.is_none()); first.schedule_diagnostic_refresh(0).await; - tokio::time::sleep(Duration::from_millis(30)).await; - assert!(client.receiver.try_recv().is_ok()); + advance(0).await; + expect_refresh(&client); } - #[tokio::test] - async fn dropping_diagnostics_cancels_pending_refresh() { + #[tokio::test(start_paused = true)] + async fn a_later_edit_restarts_the_debounce_interval() { let (diagnostic, client) = diagnostic(); - diagnostic.schedule_diagnostic_refresh(30).await; + diagnostic.schedule_diagnostic_refresh(1000).await; + advance(900).await; + diagnostic.schedule_diagnostic_refresh(1000).await; + advance(100).await; + assert!(client.receiver.try_recv().is_err()); + advance(900).await; + expect_refresh(&client); + } + + #[tokio::test(start_paused = true)] + async fn dropping_diagnostics_cancels_without_retaining_the_client() { + let (diagnostic, client) = diagnostic(); + let weak_client = Arc::downgrade(&diagnostic.client); + let references = Arc::strong_count(&diagnostic.client); + diagnostic.schedule_diagnostic_refresh(1000).await; + assert_eq!(Arc::strong_count(&diagnostic.client), references); drop(diagnostic); - tokio::time::sleep(Duration::from_millis(100)).await; + assert!(weak_client.upgrade().is_none()); + advance(1000).await; + assert!(client.receiver.try_recv().is_err()); + } + + #[tokio::test(start_paused = true)] + async fn closing_context_stops_refreshes_while_snapshots_remain_alive() { + let (server, client) = Connection::memory(); + let context = ServerContext::new(Arc::new(server), Default::default()); + let snapshot = context.snapshot(); + snapshot + .file_diagnostic() + .schedule_diagnostic_refresh(1000) + .await; + context.close().await; + // An already queued update must not restart refreshes during shutdown. + snapshot + .file_diagnostic() + .schedule_diagnostic_refresh(0) + .await; + advance(1000).await; assert!(client.receiver.try_recv().is_err()); } } diff --git a/crates/emmylua_ls/src/context/mod.rs b/crates/emmylua_ls/src/context/mod.rs index 4d6040d8a..6cb5d2e5c 100644 --- a/crates/emmylua_ls/src/context/mod.rs +++ b/crates/emmylua_ls/src/context/mod.rs @@ -53,7 +53,6 @@ impl ServerContext { client.clone(), )); let workspace_manager = Arc::new(Mutex::new(WorkspaceManager::new( - client.clone(), file_diagnostic.clone(), lsp_features.clone(), ))); @@ -136,6 +135,10 @@ impl ServerContext { } pub async fn close(&self) { + self.inner + .file_diagnostic + .shutdown_diagnostic_refresh() + .await; let mut workspace_manager = self.inner.workspace_manager.lock().await; workspace_manager.watcher = None; } diff --git a/crates/emmylua_ls/src/context/snapshot.rs b/crates/emmylua_ls/src/context/snapshot.rs index 4f01ab46b..74187cbec 100644 --- a/crates/emmylua_ls/src/context/snapshot.rs +++ b/crates/emmylua_ls/src/context/snapshot.rs @@ -30,6 +30,17 @@ impl ServerContextSnapshot { &self.inner.file_diagnostic } + /// Notify pull clients after a change that can affect other documents. + pub async fn schedule_diagnostic_refresh(&self, interval: u64) { + if self.lsp_features().supports_pull_diagnostic() + && self.lsp_features().supports_refresh_diagnostic() + { + self.file_diagnostic() + .schedule_diagnostic_refresh(interval) + .await; + } + } + pub fn workspace_manager(&self) -> &Mutex { &self.inner.workspace_manager } diff --git a/crates/emmylua_ls/src/context/workspace_manager.rs b/crates/emmylua_ls/src/context/workspace_manager.rs index eff5a8268..f419eb0e9 100644 --- a/crates/emmylua_ls/src/context/workspace_manager.rs +++ b/crates/emmylua_ls/src/context/workspace_manager.rs @@ -7,7 +7,7 @@ use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicU64, Ordering}; use std::{path::PathBuf, sync::Arc, time::Duration}; use super::workspace_state::{OpenFilesSnapshot, WorkspaceState}; -use super::{AnalysisState, ClientProxy, DiagnosticService}; +use super::{AnalysisState, DiagnosticService}; use crate::context::ServerContextSnapshot; use crate::context::lsp_features::LspFeatures; use crate::handlers::{ClientConfig, init_analysis, register_files_watch}; @@ -19,7 +19,6 @@ use tokio::sync::Mutex as AsyncMutex; use tokio_util::sync::CancellationToken; pub struct WorkspaceManager { - client: Arc, config_reload_token: Arc, reload_lock: Arc>, reload_generation: Arc, @@ -31,13 +30,8 @@ pub struct WorkspaceManager { } impl WorkspaceManager { - pub fn new( - client: Arc, - file_diagnostic: Arc, - lsp_features: Arc, - ) -> Self { + pub fn new(file_diagnostic: Arc, lsp_features: Arc) -> Self { Self { - client, config_reload_token: Arc::new(PendingTask::default()), reload_lock: Arc::new(AsyncMutex::new(())), reload_generation: Arc::new(AtomicU64::new(0)), @@ -137,7 +131,6 @@ impl WorkspaceManager { // } // let analysis = self.analysis.clone(); - // let client = self.client.clone(); // let file_diagnostic = self.file_diagnostic.clone(); // let lsp_features = self.lsp_features.clone(); // let reindex_token = self.reindex_token.clone(); @@ -161,7 +154,6 @@ impl WorkspaceManager { // refresh_workspace_diagnostics( // file_diagnostic, // lsp_features, - // client, // workspace_diagnostic_level, // ) // .await; @@ -171,7 +163,6 @@ impl WorkspaceManager { fn reload_task_handles(&self) -> ReloadTaskHandles { ReloadTaskHandles { - client: self.client.clone(), file_diagnostic: self.file_diagnostic.clone(), lsp_features: self.lsp_features.clone(), reload_lock: self.reload_lock.clone(), @@ -305,14 +296,15 @@ impl PendingTask { async fn refresh_workspace_diagnostics( file_diagnostic: Arc, lsp_features: Arc, - client: Arc, workspace_diagnostic_level: Arc, ) { file_diagnostic.cancel_workspace_diagnostic().await; workspace_diagnostic_level.store(WorkspaceDiagnosticLevel::Fast.to_u8(), Ordering::Release); if lsp_features.supports_workspace_diagnostic() { - client.refresh_workspace_diagnostics(); + if lsp_features.supports_refresh_diagnostic() { + file_diagnostic.schedule_diagnostic_refresh(0).await; + } } else { file_diagnostic .add_workspace_diagnostic_task(500, true) @@ -322,7 +314,6 @@ async fn refresh_workspace_diagnostics( #[derive(Clone)] struct ReloadTaskHandles { - client: Arc, file_diagnostic: Arc, lsp_features: Arc, reload_lock: Arc>, @@ -351,7 +342,6 @@ fn spawn_workspace_reload_task( refresh_workspace_diagnostics( handles.file_diagnostic, handles.lsp_features, - handles.client, handles.workspace_diagnostic_level, ) .await; diff --git a/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs b/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs index 2f1516197..9258833f7 100644 --- a/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs +++ b/crates/emmylua_ls/src/handlers/text_document/text_document_handler.rs @@ -1,7 +1,7 @@ use emmylua_code_analysis::{Emmyrc, uri_to_file_path}; use lsp_types::{ DidChangeTextDocumentParams, DidCloseTextDocumentParams, DidOpenTextDocumentParams, - DidSaveTextDocumentParams, + DidSaveTextDocumentParams, Uri, }; use std::sync::Arc; @@ -64,11 +64,8 @@ pub async fn process_did_open_text_document( let supports_pull = context.lsp_features().supports_pull_diagnostic(); // Schedule diagnostic task without holding any locks - if supports_pull && context.lsp_features().supports_refresh_diagnostic() && file_id.is_some() { - context - .file_diagnostic() - .schedule_diagnostic_refresh(interval) - .await; + if file_id.is_some() { + context.schedule_diagnostic_refresh(interval).await; } if !supports_pull { if let Some(file_id) = file_id { @@ -172,11 +169,8 @@ pub async fn process_did_change_text_document( let supports_pull = context.lsp_features().supports_pull_diagnostic(); // Schedule diagnostic task - if supports_pull && context.lsp_features().supports_refresh_diagnostic() && file_id.is_some() { - context - .file_diagnostic() - .schedule_diagnostic_refresh(interval) - .await; + if file_id.is_some() { + context.schedule_diagnostic_refresh(interval).await; } if !supports_pull { if let Some(file_id) = file_id { @@ -206,22 +200,11 @@ pub async fn process_did_close_document( let mut workspace = context.workspace_manager().lock().await; workspace.close_open_file(¶ms.text_document.uri); drop(workspace); - let lsp_features = context.lsp_features(); - // If the closed file no longer exists, remove it. if let Some(file_path) = uri_to_file_path(uri) && !file_path.exists() { - context - .analysis() - .update(|analysis| analysis.remove_file_by_uri(uri)) - .await; - - if !lsp_features.supports_pull_diagnostic() { - context - .file_diagnostic() - .clear_push_file_diagnostics(uri.clone()); - } + remove_closed_document(&context, uri).await; return Some(()); } @@ -235,17 +218,34 @@ pub async fn process_did_close_document( }) .unwrap_or(false); if !file_exists { - context - .analysis() - .update(|analysis| analysis.remove_file_by_uri(uri)) - .await; - - if !lsp_features.supports_pull_diagnostic() { - context - .file_diagnostic() - .clear_push_file_diagnostics(uri.clone()); - } + remove_closed_document(&context, uri).await; } Some(()) } + +/// Closing a deleted document can remove declarations used by other files. +async fn remove_closed_document(context: &ServerContextSnapshot, uri: &Uri) { + let (removed, interval) = context + .analysis() + .update(|analysis| { + let removed = analysis.get_file_id(uri).is_some(); + analysis.remove_file_by_uri(uri); + ( + removed, + analysis + .get_emmyrc() + .diagnostics + .diagnostic_interval + .unwrap_or(500), + ) + }) + .await; + if !context.lsp_features().supports_pull_diagnostic() { + context + .file_diagnostic() + .clear_push_file_diagnostics(uri.clone()); + } else if removed { + context.schedule_diagnostic_refresh(interval).await; + } +} diff --git a/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs b/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs index a3a7c0e65..d3eae0f64 100644 --- a/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs +++ b/crates/emmylua_ls/src/handlers/text_document/watched_file_handler.rs @@ -101,11 +101,8 @@ pub async fn process_did_change_watched_files( }) .await; if lsp_features.supports_pull_diagnostic() { - if (removed_file || !file_ids.is_empty()) && lsp_features.supports_refresh_diagnostic() { - context - .file_diagnostic() - .schedule_diagnostic_refresh(interval) - .await; + if removed_file || !file_ids.is_empty() { + context.schedule_diagnostic_refresh(interval).await; } } else { context diff --git a/crates/emmylua_ls/src/handlers/workspace/did_rename_files.rs b/crates/emmylua_ls/src/handlers/workspace/did_rename_files.rs index ed2324f46..ba333a7b5 100644 --- a/crates/emmylua_ls/src/handlers/workspace/did_rename_files.rs +++ b/crates/emmylua_ls/src/handlers/workspace/did_rename_files.rs @@ -63,10 +63,16 @@ pub async fn process_did_rename_files_handler( // If there are renamed files, prompt the user whether to update require paths. if !all_renames.is_empty() { - let encoding = context + let (encoding, interval) = context .analysis() - .with_snapshot(|analysis| analysis.get_emmyrc().workspace.encoding.clone()) - .unwrap_or_default(); + .with_snapshot(|analysis| { + let config = analysis.get_emmyrc(); + ( + config.workspace.encoding.clone(), + config.diagnostics.diagnostic_interval.unwrap_or(500), + ) + }) + .unwrap_or_else(|| (String::new(), 500)); context .analysis() .update(|analysis| { @@ -81,6 +87,8 @@ pub async fn process_did_rename_files_handler( }) .await; + context.schedule_diagnostic_refresh(interval).await; + // try_modify_require_path only needs to be computed synchronously on a snapshot; release it immediately and then await the user prompt. let changes = context .analysis() diff --git a/crates/emmylua_ls/src/server/message_processor.rs b/crates/emmylua_ls/src/server/message_processor.rs index 491408527..fad0e73b3 100644 --- a/crates/emmylua_ls/src/server/message_processor.rs +++ b/crates/emmylua_ls/src/server/message_processor.rs @@ -95,6 +95,14 @@ impl ServerMessageProcessor { ) -> Result> { match msg { Message::Request(req) => { + if req.method == "shutdown" { + // Stop refreshes before acknowledging shutdown and waiting for exit. + server_context + .snapshot() + .file_diagnostic() + .shutdown_diagnostic_refresh() + .await; + } if connection.handle_shutdown(&req).await? { server_context.close().await; return Ok(true); // Shutdown requested diff --git a/crates/emmylua_ls/tests/diagnostic_refresh.py b/crates/emmylua_ls/tests/diagnostic_refresh.py index 198fc895a..42bacfe83 100644 --- a/crates/emmylua_ls/tests/diagnostic_refresh.py +++ b/crates/emmylua_ls/tests/diagnostic_refresh.py @@ -1,9 +1,12 @@ -"""Protocol regression: python3 crates/emmylua_ls/tests/diagnostic_refresh.py target/release/emmylua_ls. +"""Protocol regression: python3 crates/emmylua_ls/tests/diagnostic_refresh.py . Uses only the standard library. Fixtures and server logs stay under build/. +Debounce counts/timing are tested with a paused clock in Rust; this test waits +for analysis changes and refresh requests instead of assuming a processing speed. """ import json import queue +import shutil import subprocess import sys import tempfile @@ -11,37 +14,60 @@ import time from pathlib import Path +TIMEOUT = 20 +INTERVAL = 100 +REQUIRED = '---@class Provider\n---@field set_values fun()\n' +OPTIONAL = '---@class Provider\n-- ---@field set_values fun()\n' +CONSUMER = '---@type Provider\nlocal settings_table = {}\nreturn settings_table\n' + class Client: - def __init__(self, binary, root, refresh): + def __init__(self, binary, root, refresh, pull=True): self.root = root self.messages = queue.Queue() self.next_id = 0 + self.refresh_count = 0 + self.configuration = {} self.log = (root / 'server.log').open('w') - self.process = subprocess.Popen([str(binary)], cwd=root, stdin=subprocess.PIPE, - stdout=subprocess.PIPE, stderr=self.log) + self.process = subprocess.Popen( + [str(binary), '--log-path', str(root / 'logs'), + '--resources-path', str(root.parent / 'diagnostic-refresh-resources')], + cwd=root, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self.log) threading.Thread(target=self.read, daemon=True).start() - result = self.request('initialize', { - 'processId': None, 'rootUri': root.as_uri(), - 'workspaceFolders': [{'uri': root.as_uri(), 'name': 'test'}], - 'capabilities': {'textDocument': {'diagnostic': {}}, - 'workspace': {'diagnostics': {'refreshSupport': refresh}}}}) - assert result['capabilities']['diagnosticProvider']['interFileDependencies'] - self.send('initialized', {}) - time.sleep(1) + try: + result = self.request('initialize', { + 'processId': None, 'rootUri': root.as_uri(), + 'workspaceFolders': [{'uri': root.as_uri(), 'name': 'test'}], + 'capabilities': { + 'textDocument': {'diagnostic': {}} if pull else {}, + 'workspace': { + 'configuration': True, + # Use only explicitly sent events, avoiding duplicate OS watcher events. + 'didChangeWatchedFiles': {'dynamicRegistration': True}, + 'diagnostics': {'refreshSupport': refresh}}}}) + assert result['capabilities']['diagnosticProvider']['interFileDependencies'] + self.send('initialized', {}) + # This response is dispatched only after the server finishes initialization. + self.diagnostic() + except BaseException: + self.close() + raise def read(self): - while True: - headers = {} + try: while True: - line = self.process.stdout.readline() - if not line: - return - if line == b'\r\n': - break - key, value = line.decode().split(':', 1) - headers[key.lower()] = value.strip() - self.messages.put(json.loads(self.process.stdout.read(int(headers['content-length'])))) + headers = {} + while True: + line = self.process.stdout.readline() + if not line: + raise EOFError('Language server closed stdout') + if line == b'\r\n': + break + key, value = line.decode().split(':', 1) + headers[key.lower()] = value.strip() + self.messages.put(json.loads(self.process.stdout.read(int(headers['content-length'])))) + except Exception as error: + self.messages.put(error) def write(self, message): raw = json.dumps({'jsonrpc': '2.0', **message}).encode() @@ -53,80 +79,163 @@ def send(self, method, params): def receive(self, timeout): message = self.messages.get(timeout=timeout) + if isinstance(message, Exception): + raise message if 'method' in message and 'id' in message: - result = [] if message['method'] == 'workspace/configuration' else None + method = message['method'] + if method == 'workspace/configuration': + result = [self.configuration if item.get('section') == 'emmylua' else None + for item in message['params']['items']] + elif method == 'workspace/diagnostic/refresh': + self.refresh_count += 1 + result = None + elif method in ('client/registerCapability', 'client/unregisterCapability', + 'window/workDoneProgress/create', 'window/showMessageRequest'): + # Registration/progress results are null; null declines a rename prompt. + result = None + else: + self.write({'id': message['id'], 'error': { + 'code': -32601, 'message': f'Unsupported test-client request: {method}'}}) + raise AssertionError(f'Unexpected server request: {message}') self.write({'id': message['id'], 'result': result}) return message def request(self, method, params): self.next_id += 1 self.write({'id': self.next_id, 'method': method, 'params': params}) - deadline = time.monotonic() + 15 + deadline = time.monotonic() + TIMEOUT while time.monotonic() < deadline: message = self.receive(max(0.01, deadline - time.monotonic())) if message.get('id') == self.next_id and 'method' not in message: assert 'error' not in message, message return message['result'] - raise AssertionError('Response timed out') + raise AssertionError(f'Response timed out: {method}') - def refreshes(self, duration=0.6): - count = 0 + def pump_for(self, duration): deadline = time.monotonic() + duration while time.monotonic() < deadline: try: - message = self.receive(max(0.01, deadline - time.monotonic())) + self.receive(max(0.01, deadline - time.monotonic())) except queue.Empty: break - count += message.get('method') == 'workspace/diagnostic/refresh' - return count + + def expect_refresh(self, before, expected): + if expected: + deadline = time.monotonic() + TIMEOUT + while self.refresh_count == before: + self.receive(max(0.01, deadline - time.monotonic())) + assert time.monotonic() < deadline, 'Diagnostic refresh timed out' + # Observe absence only after the expected analysis state has been reached. + # Also drain trailing refreshes before the next operation; exact debounce + # counts are deliberately left to the paused-clock Rust tests. + self.pump_for(INTERVAL / 1000 * 3) + assert (self.refresh_count > before) == expected, 'Unexpected refresh capability behavior' def diagnostic(self): return self.request('textDocument/diagnostic', { 'textDocument': {'uri': (self.root / 'settings.lua').as_uri()}})['items'] + def wait_missing(self, missing): + deadline = time.monotonic() + TIMEOUT + while True: + diagnostics = self.diagnostic() + if any(d.get('code') == 'missing-fields' for d in diagnostics) == missing: + return + assert time.monotonic() < deadline, diagnostics + self.pump_for(0.05) + + def changed(self, notify, missing, refresh): + before = self.refresh_count + notify() + self.wait_missing(missing) + self.expect_refresh(before, refresh) + + def open(self, path, text): + self.send('textDocument/didOpen', {'textDocument': { + 'uri': path.as_uri(), 'languageId': 'lua', 'version': 1, 'text': text}}) + + def watch(self, path, kind): + self.send('workspace/didChangeWatchedFiles', { + 'changes': [{'uri': path.as_uri(), 'type': kind}]}) + def close(self): - self.process.terminate() - self.process.wait(timeout=5) + if self.process.poll() is None: + self.process.terminate() + try: + self.process.wait(timeout=5) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait(timeout=5) self.process.stdin.close() self.process.stdout.close() self.log.close() -def run(binary, root, refresh): - required = '---@class Provider\n---@field set_values fun()\n' - optional = '---@class Provider\n-- ---@field set_values fun()\n' - consumer = '---@type Provider\nlocal settings_table = {}\nreturn settings_table\n' - (root / 'types.lua').write_text(required) - (root / 'settings.lua').write_text(consumer) - (root / '.emmyrc.json').write_text('{"diagnostics":{"diagnosticInterval":100}}') - client = Client(binary, root, refresh) +def run(binary, root, refresh, pull): + types = root / 'types.lua' + config = root / '.emmyrc.json' + types.write_text(OPTIONAL) + (root / 'settings.lua').write_text(CONSUMER) + config.write_text(json.dumps({'diagnostics': {'diagnosticInterval': INTERVAL}})) + client = Client(binary, root, refresh, pull) + expected = refresh and pull try: - for name, text in [('types.lua', required), ('settings.lua', consumer)]: - client.send('textDocument/didOpen', {'textDocument': { - 'uri': (root / name).as_uri(), 'languageId': 'lua', 'version': 1, 'text': text}}) - client.refreshes() - assert any(d.get('code') == 'missing-fields' for d in client.diagnostic()) - version = 1 - for text, missing in [(optional, False), (required, True)]: - for _ in range(10): - version += 1 - client.send('textDocument/didChange', {'textDocument': { - 'uri': (root / 'types.lua').as_uri(), 'version': version}, - 'contentChanges': [{'text': text}]}) - assert client.refreshes() == int(refresh), 'Edits must coalesce and respect capabilities' - assert any(d.get('code') == 'missing-fields' for d in client.diagnostic()) == missing - client.send('textDocument/didClose', {'textDocument': {'uri': (root / 'types.lua').as_uri()}}) - for kind, text, missing in [(2, optional, False), (2, required, True), - (3, None, False), (1, required, True)]: - path = root / 'types.lua' + # Different unsaved contents prove that the open queue has been processed. + client.open(root / 'settings.lua', CONSUMER) + client.changed(lambda: client.open(types, REQUIRED), True, expected) + for version, text, missing in [(2, OPTIONAL, False), (3, REQUIRED, True)]: + client.changed(lambda: client.send('textDocument/didChange', { + 'textDocument': {'uri': types.as_uri(), 'version': version}, + 'contentChanges': [{'text': text}]}), missing, expected) + + # Closing a deleted file removes its declarations without a watcher event. + types.unlink() + client.changed(lambda: client.send('textDocument/didClose', { + 'textDocument': {'uri': types.as_uri()}}), False, expected) + for kind, text, missing in [(1, REQUIRED, True), (2, OPTIONAL, False), + (2, REQUIRED, True), (3, None, False), (1, REQUIRED, True)]: if text is None: - path.unlink() + types.unlink() else: - path.write_text(text) - client.send('workspace/didChangeWatchedFiles', { - 'changes': [{'uri': path.as_uri(), 'type': kind}]}) - assert client.refreshes() == int(refresh), 'Watcher refresh must respect capabilities' - assert any(d.get('code') == 'missing-fields' for d in client.diagnostic()) == missing + types.write_text(text) + client.changed(lambda: client.watch(types, kind), missing, expected) + + # An untracked deletion changes no indexed declarations. + before = client.refresh_count + client.watch(root / 'never-indexed.lua', 3) + client.wait_missing(True) + client.expect_refresh(before, False) + + # Rename updates analysis even when there are no require edits to prompt for. + renamed = root / 'renamed.lua' + types.rename(renamed) + renamed.write_text(OPTIONAL) + client.changed(lambda: client.send('workspace/didRenameFiles', { + 'files': [{'oldUri': types.as_uri(), 'newUri': renamed.as_uri()}]}), False, expected) + renamed.write_text(REQUIRED) + client.changed(lambda: client.watch(renamed, 2), True, expected) + + # Watched config reloads and client configuration changes both refresh, + # and both must honor clients that do not support refresh requests. + for disabled in [True, False]: + config.write_text(json.dumps({'diagnostics': { + 'diagnosticInterval': INTERVAL, 'disable': ['missing-fields'] if disabled else []}})) + client.changed(lambda: client.watch(config, 2), not disabled, expected) + for disabled in [True, False]: + client.configuration = {'diagnostics': {'disable': ['missing-fields'] if disabled else []}} + client.changed(lambda: client.send('workspace/didChangeConfiguration', {'settings': {}}), + not disabled, expected) + + # Leave the process alive after shutdown: no refresh may arrive between + # its acknowledgement and exit, even if updates were queued just before it. + client.open(renamed, OPTIONAL) + client.request('shutdown', None) + at_shutdown = client.refresh_count + client.pump_for(INTERVAL / 1000 * 3) + assert client.refresh_count == at_shutdown, 'Refresh received after shutdown response' + client.send('exit', {}) + client.process.stdin.close() + # close() also bounds teardown if the existing stdio workers retain the connection. finally: client.close() @@ -134,7 +243,13 @@ def run(binary, root, refresh): if __name__ == '__main__': binary = Path(sys.argv[1]).resolve() Path('build').mkdir(exist_ok=True) - for refresh in (True, False): + for refresh, pull in [(True, True), (False, True), (True, False)]: with tempfile.TemporaryDirectory(prefix='diagnostic-refresh-', dir=Path('build').resolve()) as root: - run(binary, Path(root), refresh) - print('PASS: remove/restore required field, debounce, watcher change/delete/create, capability gating') + try: + run(binary, Path(root), refresh, pull) + except BaseException: + evidence = Path('build') / f'diagnostic-refresh-failure-{refresh}-{pull}' + shutil.copytree(root, evidence, dirs_exist_ok=True) + print(f'Failure evidence: {evidence}', file=sys.stderr) + raise + print('PASS: edits, close/delete, watcher events, rename, configuration reload, capability gating, shutdown')