From 4b9b14a0342bed100ba3ae2d618b2215358c7656 Mon Sep 17 00:00:00 2001 From: "Andrei G." Date: Sat, 25 Jul 2026 00:26:04 +0200 Subject: [PATCH 1/3] fix(bridge): split NotificationCache out of Translator lock diagnostics_pump wrote incoming notifications into the cache through Arc>, the same lock held by every MCP tool call. While a slow textDocument/diagnostic round-trip held that lock, the pump blocked trying to cache incoming publishDiagnostics/log/message notifications, and the LSP transport's non-blocking try_send silently dropped them once the notification channel filled. Cache-only reads (get_cached_diagnostics, read_resource, subscribe) also locked the translator purely to validate a path against immutable workspace roots, so they queued behind the same in-flight round-trip. Extract NotificationCache into its own Arc>, independent of Translator, and validate workspace-relative paths against a lock-free Arc<[PathBuf]> snapshot instead of the translator lock. get_diagnostics keeps holding the translator lock across its LSP round-trip; nothing else needs to anymore. Closes #104 --- CHANGELOG.md | 2 + crates/mcpls-core/src/bridge/mod.rs | 1 + crates/mcpls-core/src/bridge/translator.rs | 390 +++++++++------------ crates/mcpls-core/src/lib.rs | 142 ++++++-- crates/mcpls-core/src/mcp/handlers.rs | 29 +- crates/mcpls-core/src/mcp/server.rs | 66 ++-- crates/mcpls-core/src/transport.rs | 14 +- 7 files changed, 362 insertions(+), 282 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c5d50ae..4e09910 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,11 +11,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Stale document tracker on external file changes** — `DocumentTracker::ensure_open` now stats the file on every call and resyncs the LSP server via a single full-replacement `textDocument/didChange` when the on-disk content changed. Previously a file was only ever read once per session, so external edits (git checkout/stash, formatters, the MCP host's own Edit/Write tools) went unnoticed by mcpls and produced stale hover/diagnostics/completion results until the process restarted. (#102) - **CodeQL fork PR checkout** — `actions/checkout@v7` refuses to check out a fork PR's head SHA in `pull_request_target` workflows unless explicitly opted in; set `allow-unsafe-pr-checkout: true` on the CodeQL workflow's checkout step, which was failing every fork-originated PR since the checkout v6 → v7 bump. Safe here because the job carries no secrets and permissions are scoped to `security-events: write` + `contents: read` only. +- **Notification loss and cached-read stalls under translator-lock contention** — `NotificationCache` is now held behind its own `Arc>`, independent of `Arc>`, and workspace-root path validation for cache-only reads uses a lock-free `Arc<[PathBuf]>` snapshot instead of locking the translator. Previously, `diagnostics_pump` wrote into the cache through the translator lock: while that lock was held elsewhere for the duration of a slow `textDocument/diagnostic` round-trip, incoming `publishDiagnostics`/log/message notifications were silently dropped (the LSP transport forwards them via a non-blocking `try_send`), and `get_cached_diagnostics`/`read_resource`/`subscribe` — which only ever needed a cached read or a path check — queued behind that same round-trip for as long as it took to complete. (#104) ### Changed - **`DocumentState` API** — Breaking change: `DocumentState` gains a `disk: Option` field tracking the filesystem snapshot behind the resync mechanism above; existing struct-literal construction must add this field. Acceptable pre-1.0. (#102) - **`rmcp` 2.2.0** — Breaking change upstream: bump `rmcp` from 1.8.0 to 2.2.0 to align with the MCP 2025-11-25 spec. `rmcp::model::RawResource` and the `Annotated` wrapper were merged into a single flat `rmcp::model::Resource` struct; `McplsServer::list_resources` updated accordingly. +- **`McplsServer::new` / `HandlerContext::new` signatures** — Breaking change: both now take an additional `Arc>` and an `Arc<[PathBuf]>` workspace-roots snapshot, alongside the existing translator and subscriptions. `Translator::handle_cached_diagnostics` now takes `&[PathBuf]` and `&NotificationCache` instead of reading its own field; `Translator::handle_server_logs`/`handle_server_messages` became associated functions taking `&NotificationCache` directly, since neither needed translator state. Acceptable pre-1.0. (#104) ### Fixed diff --git a/crates/mcpls-core/src/bridge/mod.rs b/crates/mcpls-core/src/bridge/mod.rs index e6830bb..5acd0e6 100644 --- a/crates/mcpls-core/src/bridge/mod.rs +++ b/crates/mcpls-core/src/bridge/mod.rs @@ -15,6 +15,7 @@ pub use notifications::{ }; pub use resources::ResourceSubscriptions; pub use state::{DocumentState, DocumentTracker, path_to_uri, uri_to_path}; +pub(crate) use translator::validate_path_against_roots; pub use translator::{ Completion, CompletionsResult, DefinitionResult, Diagnostic, DiagnosticSeverity, DiagnosticsResult, DocumentChanges, DocumentSymbolsResult, FormatDocumentResult, HoverResult, diff --git a/crates/mcpls-core/src/bridge/translator.rs b/crates/mcpls-core/src/bridge/translator.rs index 958615c..76e34e1 100644 --- a/crates/mcpls-core/src/bridge/translator.rs +++ b/crates/mcpls-core/src/bridge/translator.rs @@ -33,8 +33,6 @@ pub struct Translator { lsp_servers: HashMap, /// Document state tracker. document_tracker: DocumentTracker, - /// Notification cache for LSP server notifications. - notification_cache: NotificationCache, /// Allowed workspace roots for path validation. workspace_roots: Vec, /// Custom file extension to language ID mappings. @@ -53,7 +51,6 @@ impl Translator { lsp_clients: HashMap::new(), lsp_servers: HashMap::new(), document_tracker: DocumentTracker::new(ResourceLimits::default(), HashMap::new()), - notification_cache: NotificationCache::new(), workspace_roots: vec![], extension_map: HashMap::new(), expected_languages: HashSet::new(), @@ -109,17 +106,6 @@ impl Translator { &mut self.document_tracker } - /// Get the notification cache. - #[must_use] - pub const fn notification_cache(&self) -> &NotificationCache { - &self.notification_cache - } - - /// Get a mutable reference to the notification cache. - pub const fn notification_cache_mut(&mut self) -> &mut NotificationCache { - &mut self.notification_cache - } - // TODO: These methods will be implemented in Phase 3-5 // Initialize and shutdown are now handled by LspServer in lifecycle.rs @@ -539,6 +525,40 @@ const MAX_POSITION_VALUE: u32 = 1_000_000; /// Maximum allowed range size in lines. const MAX_RANGE_LINES: u32 = 10_000; +/// Validate that `path` is within one of `workspace_roots`. +/// +/// Free function (rather than a `Translator` method) so callers that only need +/// path validation — e.g. cache-only MCP handlers — can validate against a +/// cloned, lock-free snapshot of the workspace roots instead of locking the +/// full `Arc>`, which may be held elsewhere across a slow +/// in-flight LSP round-trip. +/// +/// # Errors +/// +/// Returns `Error::PathOutsideWorkspace` if the path is outside all workspace roots. +pub fn validate_path_against_roots(path: &Path, workspace_roots: &[PathBuf]) -> Result { + let canonical = path.canonicalize().map_err(|e| Error::FileIo { + path: path.to_path_buf(), + source: e, + })?; + + // If no workspace roots configured, allow any path (backward compatibility) + if workspace_roots.is_empty() { + return Ok(canonical); + } + + // Check if path is within any workspace root + for root in workspace_roots { + if let Ok(canonical_root) = root.canonicalize() + && canonical.starts_with(&canonical_root) + { + return Ok(canonical); + } + } + + Err(Error::PathOutsideWorkspace(path.to_path_buf())) +} + impl Translator { /// Validate that a path is within allowed workspace boundaries. /// @@ -546,26 +566,7 @@ impl Translator { /// /// Returns `Error::PathOutsideWorkspace` if the path is outside all workspace roots. pub(crate) fn validate_path(&self, path: &Path) -> Result { - let canonical = path.canonicalize().map_err(|e| Error::FileIo { - path: path.to_path_buf(), - source: e, - })?; - - // If no workspace roots configured, allow any path (backward compatibility) - if self.workspace_roots.is_empty() { - return Ok(canonical); - } - - // Check if path is within any workspace root - for root in &self.workspace_roots { - if let Ok(canonical_root) = root.canonicalize() - && canonical.starts_with(&canonical_root) - { - return Ok(canonical); - } - } - - Err(Error::PathOutsideWorkspace(path.to_path_buf())) + validate_path_against_roots(path, &self.workspace_roots) } /// Get a cloned LSP client for a file path based on language detection. @@ -1473,49 +1474,54 @@ impl Translator { /// Handle cached diagnostics request. /// + /// Associated function rather than a method: it reads the caller-supplied + /// `workspace_roots` and `cache` instead of `self`, so callers that only + /// need a cached read (e.g. the `get_cached_diagnostics` MCP tool) never + /// need to lock `Arc>`, which may be held elsewhere + /// across a slow in-flight LSP round-trip. + /// /// # Errors /// /// Returns an error if the path is invalid or outside workspace boundaries. - pub fn handle_cached_diagnostics(&mut self, file_path: &str) -> Result { + pub fn handle_cached_diagnostics( + workspace_roots: &[PathBuf], + file_path: &str, + cache: &NotificationCache, + ) -> Result { let path = PathBuf::from(file_path); - let validated_path = self.validate_path(&path)?; + let validated_path = validate_path_against_roots(&path, workspace_roots)?; // Use path_to_uri (strips \\?\ on Windows) so the key matches what // rust-analyzer stores in publishDiagnostics notifications. let uri = path_to_uri(&validated_path).to_string(); - let diagnostics = - self.notification_cache - .get_diagnostics(&uri) - .map_or_else(Vec::new, |diag_info| { - diag_info - .diagnostics - .iter() - .map(|diag| Diagnostic { - range: normalize_range(diag.range), - severity: match diag.severity { - Some(lsp_types::DiagnosticSeverity::ERROR) => { - DiagnosticSeverity::Error - } - Some(lsp_types::DiagnosticSeverity::WARNING) => { - DiagnosticSeverity::Warning - } - Some(lsp_types::DiagnosticSeverity::INFORMATION) => { - DiagnosticSeverity::Information - } - Some(lsp_types::DiagnosticSeverity::HINT) => { - DiagnosticSeverity::Hint - } - _ => DiagnosticSeverity::Information, - }, - message: diag.message.clone(), - code: diag.code.as_ref().map(|c| match c { - lsp_types::NumberOrString::Number(n) => n.to_string(), - lsp_types::NumberOrString::String(s) => s.clone(), - }), - }) - .collect() - }); + let diagnostics = cache + .get_diagnostics(&uri) + .map_or_else(Vec::new, |diag_info| { + diag_info + .diagnostics + .iter() + .map(|diag| Diagnostic { + range: normalize_range(diag.range), + severity: match diag.severity { + Some(lsp_types::DiagnosticSeverity::ERROR) => DiagnosticSeverity::Error, + Some(lsp_types::DiagnosticSeverity::WARNING) => { + DiagnosticSeverity::Warning + } + Some(lsp_types::DiagnosticSeverity::INFORMATION) => { + DiagnosticSeverity::Information + } + Some(lsp_types::DiagnosticSeverity::HINT) => DiagnosticSeverity::Hint, + _ => DiagnosticSeverity::Information, + }, + message: diag.message.clone(), + code: diag.code.as_ref().map(|c| match c { + lsp_types::NumberOrString::Number(n) => n.to_string(), + lsp_types::NumberOrString::String(s) => s.clone(), + }), + }) + .collect() + }); Ok(DiagnosticsResult { diagnostics }) } @@ -1526,7 +1532,7 @@ impl Translator { /// /// Returns an error if the `min_level` parameter is invalid. pub fn handle_server_logs( - &mut self, + cache: &NotificationCache, limit: usize, min_level: Option, ) -> Result { @@ -1549,7 +1555,7 @@ impl Translator { None }; - let all_logs = self.notification_cache.get_logs(); + let all_logs = cache.get_logs(); let logs: Vec<_> = all_logs .iter() @@ -1573,8 +1579,11 @@ impl Translator { /// # Errors /// /// This method does not return errors. - pub fn handle_server_messages(&mut self, limit: usize) -> Result { - let all_messages = self.notification_cache.get_messages(); + pub fn handle_server_messages( + cache: &NotificationCache, + limit: usize, + ) -> Result { + let all_messages = cache.get_messages(); let messages: Vec<_> = all_messages.iter().take(limit).cloned().collect(); Ok(ServerMessagesResult { messages }) } @@ -2727,12 +2736,13 @@ mod tests { #[test] fn test_handle_cached_diagnostics_empty() { - let mut translator = Translator::new(); + let cache = NotificationCache::new(); let temp_dir = TempDir::new().unwrap(); let test_file = temp_dir.path().join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); - let result = translator.handle_cached_diagnostics(test_file.to_str().unwrap()); + let result = + Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); assert!(result.is_ok()); let diags = result.unwrap(); assert_eq!(diags.diagnostics.len(), 0); @@ -2742,49 +2752,41 @@ mod tests { fn test_handle_server_logs_with_filter() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); // Add some logs - translator - .notification_cache_mut() - .store_log(LogLevel::Error, "error msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Warning, "warning msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Info, "info msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Debug, "debug msg".to_string()); + cache.store_log(LogLevel::Error, "error msg".to_string()); + cache.store_log(LogLevel::Warning, "warning msg".to_string()); + cache.store_log(LogLevel::Info, "info msg".to_string()); + cache.store_log(LogLevel::Debug, "debug msg".to_string()); // Test with error filter - let result = translator.handle_server_logs(10, Some("error".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("error".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 1); assert_eq!(logs.logs[0].message, "error msg"); // Test with warning filter (includes error and warning) - let result = translator.handle_server_logs(10, Some("warning".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("warning".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 2); // Test with info filter (excludes debug) - let result = translator.handle_server_logs(10, Some("info".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("info".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 3); // Test with debug filter (includes all) - let result = translator.handle_server_logs(10, Some("debug".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("debug".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 4); // Test with invalid filter - let result = translator.handle_server_logs(10, Some("invalid".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("invalid".to_string())); assert!(matches!(result, Err(Error::InvalidToolParams(_)))); } @@ -2792,17 +2794,15 @@ mod tests { fn test_handle_server_messages_limit() { use crate::bridge::notifications::MessageType; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); // Add some messages for i in 0..10 { - translator - .notification_cache_mut() - .store_message(MessageType::Info, format!("message {i}")); + cache.store_message(MessageType::Info, format!("message {i}")); } // Test limit - let result = translator.handle_server_messages(5); + let result = Translator::handle_server_messages(&cache, 5); assert!(result.is_ok()); let messages = result.unwrap(); assert_eq!(messages.messages.len(), 5); @@ -2810,7 +2810,7 @@ mod tests { assert_eq!(messages.messages[4].message, "message 4"); // Test limit larger than available - let result = translator.handle_server_messages(100); + let result = Translator::handle_server_messages(&cache, 100); assert!(result.is_ok()); let messages = result.unwrap(); assert_eq!(messages.messages.len(), 10); @@ -2818,7 +2818,7 @@ mod tests { #[test] fn test_handle_cached_diagnostics_with_data() { - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); let temp_dir = TempDir::new().unwrap(); let test_file = temp_dir.path().join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); @@ -2850,11 +2850,10 @@ mod tests { data: None, }; - translator - .notification_cache_mut() - .store_diagnostics(&uri, Some(1), vec![diagnostic]); + cache.store_diagnostics(&uri, Some(1), vec![diagnostic]); - let result = translator.handle_cached_diagnostics(test_file.to_str().unwrap()); + let result = + Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); assert!(result.is_ok()); let diags = result.unwrap(); assert_eq!(diags.diagnostics.len(), 1); @@ -2871,7 +2870,7 @@ mod tests { #[test] #[allow(clippy::too_many_lines)] fn test_handle_cached_diagnostics_multiple_severities() { - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); let temp_dir = TempDir::new().unwrap(); let test_file = temp_dir.path().join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); @@ -2965,11 +2964,10 @@ mod tests { }, ]; - translator - .notification_cache_mut() - .store_diagnostics(&uri, Some(1), diagnostics); + cache.store_diagnostics(&uri, Some(1), diagnostics); - let result = translator.handle_cached_diagnostics(test_file.to_str().unwrap()); + let result = + Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); assert!(result.is_ok()); let diags = result.unwrap(); assert_eq!(diags.diagnostics.len(), 4); @@ -2993,7 +2991,7 @@ mod tests { #[test] fn test_handle_cached_diagnostics_with_numeric_code() { - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); let temp_dir = TempDir::new().unwrap(); let test_file = temp_dir.path().join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); @@ -3025,11 +3023,10 @@ mod tests { data: None, }; - translator - .notification_cache_mut() - .store_diagnostics(&uri, Some(1), vec![diagnostic]); + cache.store_diagnostics(&uri, Some(1), vec![diagnostic]); - let result = translator.handle_cached_diagnostics(test_file.to_str().unwrap()); + let result = + Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); assert!(result.is_ok()); let diags = result.unwrap(); assert_eq!(diags.diagnostics.len(), 1); @@ -3038,8 +3035,9 @@ mod tests { #[test] fn test_handle_cached_diagnostics_invalid_path() { - let mut translator = Translator::new(); - let result = translator.handle_cached_diagnostics("/nonexistent/path/file.rs"); + let cache = NotificationCache::new(); + let result = + Translator::handle_cached_diagnostics(&[], "/nonexistent/path/file.rs", &cache); assert!(matches!(result, Err(Error::FileIo { .. }))); } @@ -3047,22 +3045,14 @@ mod tests { fn test_handle_server_logs_no_filter() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); + + cache.store_log(LogLevel::Error, "error msg".to_string()); + cache.store_log(LogLevel::Warning, "warning msg".to_string()); + cache.store_log(LogLevel::Info, "info msg".to_string()); + cache.store_log(LogLevel::Debug, "debug msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Error, "error msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Warning, "warning msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Info, "info msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Debug, "debug msg".to_string()); - - let result = translator.handle_server_logs(10, None); + let result = Translator::handle_server_logs(&cache, 10, None); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 4); @@ -3072,19 +3062,13 @@ mod tests { fn test_handle_server_logs_error_filter_strict() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); + + cache.store_log(LogLevel::Error, "error msg".to_string()); + cache.store_log(LogLevel::Warning, "warning msg".to_string()); + cache.store_log(LogLevel::Info, "info msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Error, "error msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Warning, "warning msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Info, "info msg".to_string()); - - let result = translator.handle_server_logs(10, Some("error".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("error".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 1); @@ -3095,19 +3079,13 @@ mod tests { fn test_handle_server_logs_warning_filter_includes_errors() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); + + cache.store_log(LogLevel::Error, "error msg".to_string()); + cache.store_log(LogLevel::Warning, "warning msg".to_string()); + cache.store_log(LogLevel::Info, "info msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Error, "error msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Warning, "warning msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Info, "info msg".to_string()); - - let result = translator.handle_server_logs(10, Some("warning".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("warning".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 2); @@ -3117,19 +3095,13 @@ mod tests { fn test_handle_server_logs_info_filter_excludes_debug() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); + + cache.store_log(LogLevel::Error, "error msg".to_string()); + cache.store_log(LogLevel::Info, "info msg".to_string()); + cache.store_log(LogLevel::Debug, "debug msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Error, "error msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Info, "info msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Debug, "debug msg".to_string()); - - let result = translator.handle_server_logs(10, Some("info".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("info".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 2); @@ -3139,22 +3111,14 @@ mod tests { fn test_handle_server_logs_debug_filter_includes_all() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); + + cache.store_log(LogLevel::Error, "error msg".to_string()); + cache.store_log(LogLevel::Warning, "warning msg".to_string()); + cache.store_log(LogLevel::Info, "info msg".to_string()); + cache.store_log(LogLevel::Debug, "debug msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Error, "error msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Warning, "warning msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Info, "info msg".to_string()); - translator - .notification_cache_mut() - .store_log(LogLevel::Debug, "debug msg".to_string()); - - let result = translator.handle_server_logs(10, Some("debug".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("debug".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 4); @@ -3164,15 +3128,13 @@ mod tests { fn test_handle_server_logs_limit_applies_after_filter() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); for i in 0..10 { - translator - .notification_cache_mut() - .store_log(LogLevel::Error, format!("error {i}")); + cache.store_log(LogLevel::Error, format!("error {i}")); } - let result = translator.handle_server_logs(5, Some("error".to_string())); + let result = Translator::handle_server_logs(&cache, 5, Some("error".to_string())); assert!(result.is_ok()); let logs = result.unwrap(); assert_eq!(logs.logs.len(), 5); @@ -3184,27 +3146,25 @@ mod tests { fn test_handle_server_logs_case_insensitive_level() { use crate::bridge::notifications::LogLevel; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); - translator - .notification_cache_mut() - .store_log(LogLevel::Error, "error msg".to_string()); + cache.store_log(LogLevel::Error, "error msg".to_string()); - let result = translator.handle_server_logs(10, Some("ERROR".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("ERROR".to_string())); assert!(result.is_ok()); - let result = translator.handle_server_logs(10, Some("Error".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("Error".to_string())); assert!(result.is_ok()); - let result = translator.handle_server_logs(10, Some("eRrOr".to_string())); + let result = Translator::handle_server_logs(&cache, 10, Some("eRrOr".to_string())); assert!(result.is_ok()); } #[test] fn test_handle_server_messages_empty() { - let mut translator = Translator::new(); + let cache = NotificationCache::new(); - let result = translator.handle_server_messages(10); + let result = Translator::handle_server_messages(&cache, 10); assert!(result.is_ok()); let messages = result.unwrap(); assert_eq!(messages.messages.len(), 0); @@ -3214,22 +3174,14 @@ mod tests { fn test_handle_server_messages_with_different_types() { use crate::bridge::notifications::MessageType; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); - translator - .notification_cache_mut() - .store_message(MessageType::Error, "error".to_string()); - translator - .notification_cache_mut() - .store_message(MessageType::Warning, "warning".to_string()); - translator - .notification_cache_mut() - .store_message(MessageType::Info, "info".to_string()); - translator - .notification_cache_mut() - .store_message(MessageType::Log, "log".to_string()); - - let result = translator.handle_server_messages(10); + cache.store_message(MessageType::Error, "error".to_string()); + cache.store_message(MessageType::Warning, "warning".to_string()); + cache.store_message(MessageType::Info, "info".to_string()); + cache.store_message(MessageType::Log, "log".to_string()); + + let result = Translator::handle_server_messages(&cache, 10); assert!(result.is_ok()); let messages = result.unwrap(); assert_eq!(messages.messages.len(), 4); @@ -3243,13 +3195,11 @@ mod tests { fn test_handle_server_messages_zero_limit() { use crate::bridge::notifications::MessageType; - let mut translator = Translator::new(); + let mut cache = NotificationCache::new(); - translator - .notification_cache_mut() - .store_message(MessageType::Info, "test".to_string()); + cache.store_message(MessageType::Info, "test".to_string()); - let result = translator.handle_server_messages(0); + let result = Translator::handle_server_messages(&cache, 0); assert!(result.is_ok()); let messages = result.unwrap(); assert_eq!(messages.messages.len(), 0); @@ -3257,16 +3207,20 @@ mod tests { #[test] fn test_handle_cached_diagnostics_path_outside_workspace() { - let mut translator = Translator::new(); + let cache = NotificationCache::new(); let temp_dir1 = TempDir::new().unwrap(); let temp_dir2 = TempDir::new().unwrap(); - translator.set_workspace_roots(vec![temp_dir1.path().to_path_buf()]); + let workspace_roots = vec![temp_dir1.path().to_path_buf()]; let test_file = temp_dir2.path().join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); - let result = translator.handle_cached_diagnostics(test_file.to_str().unwrap()); + let result = Translator::handle_cached_diagnostics( + &workspace_roots, + test_file.to_str().unwrap(), + &cache, + ); assert!(matches!(result, Err(Error::PathOutsideWorkspace(_)))); } diff --git a/crates/mcpls-core/src/lib.rs b/crates/mcpls-core/src/lib.rs index 553fab2..1525713 100644 --- a/crates/mcpls-core/src/lib.rs +++ b/crates/mcpls-core/src/lib.rs @@ -44,7 +44,7 @@ use std::path::PathBuf; use std::sync::Arc; use bridge::resources::make_uri; -use bridge::{ResourceSubscriptions, Translator}; +use bridge::{NotificationCache, ResourceSubscriptions, Translator}; pub use config::ServerConfig; pub use error::Error; use lsp::{LspNotification, LspServer, ServerInitConfig}; @@ -72,14 +72,19 @@ use transport::run_stdio; /// - The cancellation watch fires (or the sender is dropped). /// - `notify_resource_updated` returns an error (peer disconnect / transport closed). /// -/// # Note on lock contention (TODO critic-S4) -/// All cache writes acquire `Arc>`, which is the same lock used -/// by every MCP tool call. Splitting `NotificationCache` into its own `Arc` -/// would eliminate this contention. Tracked as a P2 follow-up. +/// # Lock independence +/// Cache writes acquire only `Arc>`, a lock independent +/// of `Arc>`. MCP tool calls that hold the translator lock +/// across an in-flight LSP round-trip (e.g. `textDocument/diagnostic`) never +/// block this pump, so a `publishDiagnostics` notification arriving mid-request +/// is cached immediately instead of being silently dropped: the LSP transport +/// forwards notifications via `mpsc::Sender::try_send`, which drops on a full +/// channel rather than blocking, so a pump stalled on the translator lock +/// previously lost notifications under sustained push traffic. pub(crate) async fn diagnostics_pump( _lang: String, mut rx: tokio::sync::mpsc::Receiver, - translator: Arc>, + notification_cache: Arc>, subs: Arc, peer_cell: Arc>>, mut cancel_rx: tokio::sync::watch::Receiver, @@ -99,9 +104,8 @@ pub(crate) async fn diagnostics_pump( LspNotification::PublishDiagnostics(p) => { // Always cache unconditionally. { - let mut t = translator.lock().await; - t.notification_cache_mut() - .store_diagnostics(&p.uri, p.version, p.diagnostics); + let mut cache = notification_cache.lock().await; + cache.store_diagnostics(&p.uri, p.version, p.diagnostics); } // Fast path: skip URI construction when nothing is subscribed. @@ -133,14 +137,12 @@ pub(crate) async fn diagnostics_pump( } } LspNotification::LogMessage(m) => { - let mut t = translator.lock().await; - t.notification_cache_mut() - .store_log(m.typ.into(), m.message); + let mut cache = notification_cache.lock().await; + cache.store_log(m.typ.into(), m.message); } LspNotification::ShowMessage(m) => { - let mut t = translator.lock().await; - t.notification_cache_mut() - .store_message(m.typ.into(), m.message); + let mut cache = notification_cache.lock().await; + cache.store_message(m.typ.into(), m.message); } LspNotification::Progress { .. } | LspNotification::Other { .. } => {} } @@ -326,7 +328,16 @@ pub async fn serve_with(config: ServerConfig, transport: Transport) -> Result<() // in a background task and registered into this shared translator once ready. // Blocking the MCP handshake on LSP init makes slow servers exceed the client's // initialize-request timeout (Claude Code: ~60s) -> "Request timed out". + // Fixed for the server's lifetime: shared as a lock-free snapshot so + // cache-only handlers (e.g. `get_cached_diagnostics`, `read_resource`) can + // validate a path without locking `translator` below. + let workspace_roots_snapshot: Arc<[PathBuf]> = Arc::from(workspace_roots.clone()); + let translator = Arc::new(Mutex::new(translator)); + // Independent of `translator`: the pump only ever locks this, so it never + // contends with a request handler holding the translator lock across an + // in-flight LSP round-trip. + let notification_cache = Arc::new(Mutex::new(NotificationCache::new())); let subscriptions = Arc::new(ResourceSubscriptions::new()); // Peer cell is populated after the MCP transport is established (Phase B). let peer_cell = Arc::new(OnceCell::new()); @@ -344,6 +355,7 @@ pub async fn serve_with(config: ServerConfig, transport: Transport) -> Result<() spawn_lsp_servers_background( applicable_configs, Arc::clone(&translator), + Arc::clone(¬ification_cache), Arc::clone(&subscriptions), Arc::clone(&peer_cell), cancel_rx.clone(), @@ -351,7 +363,12 @@ pub async fn serve_with(config: ServerConfig, transport: Transport) -> Result<() } info!("Starting MCP server with rmcp..."); - let mcp_server = mcp::McplsServer::new(Arc::clone(&translator), Arc::clone(&subscriptions)); + let mcp_server = mcp::McplsServer::new( + Arc::clone(&translator), + Arc::clone(¬ification_cache), + Arc::clone(&workspace_roots_snapshot), + Arc::clone(&subscriptions), + ); info!("MCPLS server initialized successfully"); let result = match transport { @@ -383,6 +400,7 @@ pub async fn serve_with(config: ServerConfig, transport: Transport) -> Result<() fn spawn_lsp_servers_background( applicable_configs: Vec, translator: Arc>, + notification_cache: Arc>, subscriptions: Arc, peer_cell: Arc>>, cancel_rx: tokio::sync::watch::Receiver, @@ -433,7 +451,7 @@ fn spawn_lsp_servers_background( pumps.spawn(diagnostics_pump( lang, rx, - Arc::clone(&translator), + Arc::clone(¬ification_cache), Arc::clone(&subscriptions), Arc::clone(&peer_cell), cancel_rx.clone(), @@ -792,8 +810,8 @@ mod tests { use super::*; - fn make_translator() -> Arc> { - Arc::new(Mutex::new(Translator::new())) + fn make_cache() -> Arc> { + Arc::new(Mutex::new(NotificationCache::new())) } fn make_subs() -> Arc { @@ -809,7 +827,7 @@ mod tests { /// `PublishDiagnostics` is cached even when the peer is not yet connected. #[tokio::test] async fn test_pump_caches_before_peer_set() { - let translator = make_translator(); + let cache = make_cache(); let subs = make_subs(); let peer_cell = make_peer_cell(); let (tx, rx) = mpsc::channel(8); @@ -817,11 +835,11 @@ mod tests { // which makes the pump exit before processing any messages. let (_cancel_tx, cancel_rx) = watch::channel(false); - let t = Arc::clone(&translator); + let c = Arc::clone(&cache); tokio::spawn(diagnostics_pump( "rust".to_string(), rx, - t, + c, Arc::clone(&subs), Arc::clone(&peer_cell), cancel_rx, @@ -844,11 +862,8 @@ mod tests { loop { tokio::task::yield_now().await; let found = { - let guard = translator.lock().await; - guard - .notification_cache() - .get_diagnostics(uri.as_str()) - .is_some() + let guard = cache.lock().await; + guard.get_diagnostics(uri.as_str()).is_some() }; if found { return true; @@ -864,7 +879,7 @@ mod tests { /// Pump exits cleanly when the cancel watch sends `true`. #[tokio::test] async fn test_pump_exits_on_cancel() { - let translator = make_translator(); + let cache = make_cache(); let subs = make_subs(); let peer_cell = make_peer_cell(); let (_tx, rx) = mpsc::channel::(8); @@ -873,7 +888,7 @@ mod tests { let handle = tokio::spawn(diagnostics_pump( "rust".to_string(), rx, - translator, + cache, subs, peer_cell, cancel_rx, @@ -890,7 +905,7 @@ mod tests { /// Pump exits when the cancel sender is dropped (Err branch). #[tokio::test] async fn test_pump_exits_when_cancel_sender_dropped() { - let translator = make_translator(); + let cache = make_cache(); let subs = make_subs(); let peer_cell = make_peer_cell(); let (_tx, rx) = mpsc::channel::(8); @@ -899,7 +914,7 @@ mod tests { let handle = tokio::spawn(diagnostics_pump( "rust".to_string(), rx, - translator, + cache, subs, peer_cell, cancel_rx, @@ -911,5 +926,70 @@ mod tests { .expect("pump did not exit within timeout") .unwrap(); } + + /// Regression test for #104: the pump must cache a notification promptly + /// even while another task holds the translator lock for far longer than + /// any acceptable pump latency. Before the `NotificationCache` split, the + /// pump locked `Arc>` to cache diagnostics, so it would + /// have stalled here until the holder released the lock. + #[tokio::test] + async fn test_pump_makes_progress_while_translator_lock_held() { + let translator = Arc::new(Mutex::new(Translator::new())); + let cache = make_cache(); + let subs = make_subs(); + let peer_cell = make_peer_cell(); + let (tx, rx) = mpsc::channel(8); + let (_cancel_tx, cancel_rx) = watch::channel(false); + + // Simulate a slow in-flight MCP request (e.g. `pull_diagnostics`) + // holding the translator lock across an LSP round-trip. + let lock_acquired = Arc::new(tokio::sync::Notify::new()); + let notify = Arc::clone(&lock_acquired); + let holder = tokio::spawn(async move { + let _guard = translator.lock().await; + notify.notify_one(); + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + }); + lock_acquired.notified().await; + + tokio::spawn(diagnostics_pump( + "rust".to_string(), + rx, + Arc::clone(&cache), + subs, + peer_cell, + cancel_rx, + )); + + let uri: Uri = "file:///test/locked.rs".parse().unwrap(); + tx.send(LspNotification::PublishDiagnostics( + PublishDiagnosticsParams { + uri: uri.clone(), + diagnostics: vec![], + version: None, + }, + )) + .await + .unwrap(); + drop(tx); + + // Well within the 2 s translator lock hold: a translator-locking + // pump would still be blocked at this point. + tokio::time::timeout(std::time::Duration::from_millis(500), async { + loop { + { + let guard = cache.lock().await; + if guard.get_diagnostics(uri.as_str()).is_some() { + return; + } + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .expect("pump stalled behind translator lock"); + + holder.await.unwrap(); + } } } diff --git a/crates/mcpls-core/src/mcp/handlers.rs b/crates/mcpls-core/src/mcp/handlers.rs index 502d0c6..f3b0ca7 100644 --- a/crates/mcpls-core/src/mcp/handlers.rs +++ b/crates/mcpls-core/src/mcp/handlers.rs @@ -4,11 +4,12 @@ //! The actual tool implementations use the `#[tool]` macro from rmcp //! and are defined in the `server` module. +use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; -use crate::bridge::{ResourceSubscriptions, Translator}; +use crate::bridge::{NotificationCache, ResourceSubscriptions, Translator}; /// Shared context for all tool handlers. /// @@ -18,6 +19,19 @@ use crate::bridge::{ResourceSubscriptions, Translator}; pub struct HandlerContext { /// Translator for converting MCP calls to LSP requests. pub translator: Arc>, + /// Cache of pushed LSP notifications (diagnostics, logs, messages). + /// + /// Locked independently of `translator` so the `diagnostics_pump` task + /// never contends with a tool call holding the translator lock across an + /// in-flight LSP round-trip. + pub notification_cache: Arc>, + /// Workspace roots, fixed at startup and immutable thereafter. + /// + /// Shared as a lock-free snapshot so cache-only handlers (e.g. + /// `get_cached_diagnostics`, `read_resource`) can validate a path without + /// locking `translator`, which may be held elsewhere across a slow + /// in-flight LSP round-trip. + pub workspace_roots: Arc<[PathBuf]>, /// Set of resource URIs the MCP client has subscribed to. pub subscriptions: Arc, } @@ -27,10 +41,14 @@ impl HandlerContext { #[must_use] pub const fn new( translator: Arc>, + notification_cache: Arc>, + workspace_roots: Arc<[PathBuf]>, subscriptions: Arc, ) -> Self { Self { translator, + notification_cache, + workspace_roots, subscriptions, } } @@ -44,8 +62,15 @@ mod tests { #[test] fn test_handler_context_creation() { let translator = Arc::new(Mutex::new(Translator::new())); + let notification_cache = Arc::new(Mutex::new(NotificationCache::new())); + let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new()); let subscriptions = Arc::new(ResourceSubscriptions::new()); - let context = HandlerContext::new(translator, subscriptions); + let context = HandlerContext::new( + translator, + notification_cache, + workspace_roots, + subscriptions, + ); assert_eq!(Arc::strong_count(&context.translator), 1); } } diff --git a/crates/mcpls-core/src/mcp/server.rs b/crates/mcpls-core/src/mcp/server.rs index c450fee..266da9d 100644 --- a/crates/mcpls-core/src/mcp/server.rs +++ b/crates/mcpls-core/src/mcp/server.rs @@ -3,6 +3,7 @@ //! This module provides the MCP server that exposes LSP capabilities //! as MCP tools using the rmcp SDK. +use std::path::PathBuf; use std::sync::Arc; use rmcp::handler::server::wrapper::Parameters; @@ -23,7 +24,9 @@ use super::tools::{ ServerLogsParams, ServerMessagesParams, SignatureHelpParams, WorkspaceSymbolParams, }; use crate::bridge::resources::{make_uri, parse_uri}; -use crate::bridge::{ResourceSubscriptions, Translator}; +use crate::bridge::{ + NotificationCache, ResourceSubscriptions, Translator, validate_path_against_roots, +}; /// MCP server that exposes LSP capabilities as tools. #[derive(Clone)] @@ -33,13 +36,21 @@ pub struct McplsServer { #[tool_router] impl McplsServer { - /// Create a new MCP server with the given translator and subscriptions. + /// Create a new MCP server with the given translator, notification cache, + /// workspace roots, and subscriptions. #[must_use] pub fn new( translator: Arc>, + notification_cache: Arc>, + workspace_roots: Arc<[PathBuf]>, subscriptions: Arc, ) -> Self { - let context = Arc::new(HandlerContext::new(translator, subscriptions)); + let context = Arc::new(HandlerContext::new( + translator, + notification_cache, + workspace_roots, + subscriptions, + )); Self { context } } @@ -377,8 +388,8 @@ impl McplsServer { Parameters(CachedDiagnosticsParams { file_path }): Parameters, ) -> Result { let result = { - let mut translator = self.context.translator.lock().await; - translator.handle_cached_diagnostics(&file_path) + let cache = self.context.notification_cache.lock().await; + Translator::handle_cached_diagnostics(&self.context.workspace_roots, &file_path, &cache) }; match result { @@ -397,8 +408,8 @@ impl McplsServer { Parameters(ServerLogsParams { limit, min_level }): Parameters, ) -> Result { let result = { - let mut translator = self.context.translator.lock().await; - translator.handle_server_logs(limit, min_level) + let cache = self.context.notification_cache.lock().await; + Translator::handle_server_logs(&cache, limit, min_level) }; match result { @@ -417,8 +428,8 @@ impl McplsServer { Parameters(ServerMessagesParams { limit }): Parameters, ) -> Result { let result = { - let mut translator = self.context.translator.lock().await; - translator.handle_server_messages(limit) + let cache = self.context.notification_cache.lock().await; + Translator::handle_server_messages(&cache, limit) }; match result { @@ -590,12 +601,11 @@ impl ServerHandler for McplsServer { parse_uri(&request.uri).map_err(|e| McpError::invalid_params(e.to_string(), None))?; // Enforce workspace-root containment — mirrors the guard in every LSP tool. - { - let translator = self.context.translator.lock().await; - translator - .validate_path(&path) - .map_err(|e| McpError::invalid_params(e.to_string(), None))?; - } + // Validated against a lock-free snapshot of workspace_roots (fixed at + // startup) so this cache-only read never waits on the translator lock, + // which may be held elsewhere across a slow in-flight LSP round-trip. + validate_path_against_roots(&path, &self.context.workspace_roots) + .map_err(|e| McpError::invalid_params(e.to_string(), None))?; let lsp_uri = crate::bridge::path_to_uri(&path); @@ -603,11 +613,8 @@ impl ServerHandler for McplsServer { // in the response shape. Currently both return `{"diagnostics":null}` which is // ambiguous for clients that need to know whether analysis has run yet. let diagnostics = { - let translator = self.context.translator.lock().await; - translator - .notification_cache() - .get_diagnostics(lsp_uri.as_str()) - .cloned() + let cache = self.context.notification_cache.lock().await; + cache.get_diagnostics(lsp_uri.as_str()).cloned() }; let json = serde_json::to_string(&diagnostics) @@ -628,12 +635,10 @@ impl ServerHandler for McplsServer { parse_uri(&request.uri).map_err(|e| McpError::invalid_params(e.to_string(), None))?; // Enforce workspace-root containment (same invariant as every LSP tool). - { - let translator = self.context.translator.lock().await; - translator - .validate_path(&path) - .map_err(|e| McpError::invalid_params(e.to_string(), None))?; - } + // Validated against a lock-free snapshot of workspace_roots so subscribing + // never waits on the translator lock (see `read_resource`). + validate_path_against_roots(&path, &self.context.workspace_roots) + .map_err(|e| McpError::invalid_params(e.to_string(), None))?; // TODO(S3): If diagnostics are already cached for this URI, emit a synthetic // notify_resource_updated so clients subscribing after initial workspace indexing @@ -693,8 +698,15 @@ mod tests { fn create_test_server() -> McplsServer { let translator = Arc::new(Mutex::new(Translator::new())); + let notification_cache = Arc::new(Mutex::new(NotificationCache::new())); + let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new()); let subscriptions = Arc::new(ResourceSubscriptions::new()); - McplsServer::new(translator, subscriptions) + McplsServer::new( + translator, + notification_cache, + workspace_roots, + subscriptions, + ) } #[tokio::test] diff --git a/crates/mcpls-core/src/transport.rs b/crates/mcpls-core/src/transport.rs index ad5fff9..319c36b 100644 --- a/crates/mcpls-core/src/transport.rs +++ b/crates/mcpls-core/src/transport.rs @@ -246,16 +246,19 @@ mod tests { /// Verifies `run_http` binds successfully and accepts TCP connections. #[tokio::test] async fn test_run_http_binds() { + use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; - use crate::bridge::{ResourceSubscriptions, Translator}; + use crate::bridge::{NotificationCache, ResourceSubscriptions, Translator}; use crate::mcp::McplsServer; let translator = Arc::new(Mutex::new(Translator::new())); + let notification_cache = Arc::new(Mutex::new(NotificationCache::new())); + let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new()); let subs = Arc::new(ResourceSubscriptions::new()); - let server = McplsServer::new(translator, subs); + let server = McplsServer::new(translator, notification_cache, workspace_roots, subs); // Bind port 0 so the OS assigns a free port. let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -283,11 +286,12 @@ mod tests { /// Verifies `run_http` returns an error when the bind address is already in use. #[tokio::test] async fn test_run_http_bind_error() { + use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; - use crate::bridge::{ResourceSubscriptions, Translator}; + use crate::bridge::{NotificationCache, ResourceSubscriptions, Translator}; use crate::mcp::McplsServer; // Hold a listener to make the port unavailable. @@ -295,8 +299,10 @@ mod tests { let addr = occupied.local_addr().unwrap(); let translator = Arc::new(Mutex::new(Translator::new())); + let notification_cache = Arc::new(Mutex::new(NotificationCache::new())); + let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new()); let subs = Arc::new(ResourceSubscriptions::new()); - let server = McplsServer::new(translator, subs); + let server = McplsServer::new(translator, notification_cache, workspace_roots, subs); let cfg = HttpConfig { bind: addr, From 6b8669782f069cede2efbb26d368cbed26b2029b Mon Sep 17 00:00:00 2001 From: "Andrei G." Date: Sat, 25 Jul 2026 00:48:54 +0200 Subject: [PATCH 2/3] fix(bridge): address Copilot review findings on notification cache lock split get_cached_diagnostics still held the notification_cache mutex across canonicalize() and the cached-diagnostics-to-MCP-shape mapping, reintroducing the same lock-held-during-syscall class this branch exists to eliminate. Split Translator::handle_cached_diagnostics into cached_diagnostics_uri (path validation, no lock) and diagnostics_from_cache_entry (pure mapping over an already-cloned entry, no lock); the cache lock is now held only for the map lookup and clone. read_resource discarded the canonicalized path returned by validate_path_against_roots and built its cache-lookup URI from the raw input path, so paths with symlinks or `..` segments missed the cache even when tracked under their canonical form. Use the validated path for the URI, matching get_cached_diagnostics. --- CHANGELOG.md | 4 +- crates/mcpls-core/src/bridge/translator.rs | 130 ++++++++++----------- crates/mcpls-core/src/mcp/server.rs | 126 +++++++++++++++++++- 3 files changed, 186 insertions(+), 74 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e09910..58fd20d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,12 +12,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Stale document tracker on external file changes** — `DocumentTracker::ensure_open` now stats the file on every call and resyncs the LSP server via a single full-replacement `textDocument/didChange` when the on-disk content changed. Previously a file was only ever read once per session, so external edits (git checkout/stash, formatters, the MCP host's own Edit/Write tools) went unnoticed by mcpls and produced stale hover/diagnostics/completion results until the process restarted. (#102) - **CodeQL fork PR checkout** — `actions/checkout@v7` refuses to check out a fork PR's head SHA in `pull_request_target` workflows unless explicitly opted in; set `allow-unsafe-pr-checkout: true` on the CodeQL workflow's checkout step, which was failing every fork-originated PR since the checkout v6 → v7 bump. Safe here because the job carries no secrets and permissions are scoped to `security-events: write` + `contents: read` only. - **Notification loss and cached-read stalls under translator-lock contention** — `NotificationCache` is now held behind its own `Arc>`, independent of `Arc>`, and workspace-root path validation for cache-only reads uses a lock-free `Arc<[PathBuf]>` snapshot instead of locking the translator. Previously, `diagnostics_pump` wrote into the cache through the translator lock: while that lock was held elsewhere for the duration of a slow `textDocument/diagnostic` round-trip, incoming `publishDiagnostics`/log/message notifications were silently dropped (the LSP transport forwards them via a non-blocking `try_send`), and `get_cached_diagnostics`/`read_resource`/`subscribe` — which only ever needed a cached read or a path check — queued behind that same round-trip for as long as it took to complete. (#104) +- **`get_cached_diagnostics` still held the cache lock across `canonicalize()` and result mapping** — the `Translator::handle_cached_diagnostics` split above narrowed but didn't fully close the same lock-contention class: path validation (a filesystem `canonicalize()` syscall) and the cached-diagnostics-to-MCP-shape mapping still ran while holding the `notification_cache` mutex that `diagnostics_pump` also needs. Split into `Translator::cached_diagnostics_uri` (path validation, no lock involved) and `Translator::diagnostics_from_cache_entry` (pure mapping over an already-cloned entry, no lock involved); `get_cached_diagnostics` now holds the cache lock only for the `get_diagnostics(&uri).cloned()` lookup itself. (#104) +- **`read_resource` used the raw, non-canonicalized path to build its diagnostics-cache URI** — `validate_path_against_roots` returns the canonicalized path but the result was discarded, so `read_resource` looked up the cache under the client's original (possibly symlinked or `..`-containing) path instead of the canonical form `diagnostics_pump` actually stores diagnostics under, causing spurious cache misses for otherwise-valid, tracked files. Fixed to build the URI from the validated/canonical path, matching `get_cached_diagnostics`. (#104) ### Changed - **`DocumentState` API** — Breaking change: `DocumentState` gains a `disk: Option` field tracking the filesystem snapshot behind the resync mechanism above; existing struct-literal construction must add this field. Acceptable pre-1.0. (#102) - **`rmcp` 2.2.0** — Breaking change upstream: bump `rmcp` from 1.8.0 to 2.2.0 to align with the MCP 2025-11-25 spec. `rmcp::model::RawResource` and the `Annotated` wrapper were merged into a single flat `rmcp::model::Resource` struct; `McplsServer::list_resources` updated accordingly. -- **`McplsServer::new` / `HandlerContext::new` signatures** — Breaking change: both now take an additional `Arc>` and an `Arc<[PathBuf]>` workspace-roots snapshot, alongside the existing translator and subscriptions. `Translator::handle_cached_diagnostics` now takes `&[PathBuf]` and `&NotificationCache` instead of reading its own field; `Translator::handle_server_logs`/`handle_server_messages` became associated functions taking `&NotificationCache` directly, since neither needed translator state. Acceptable pre-1.0. (#104) +- **`McplsServer::new` / `HandlerContext::new` signatures** — Breaking change: both now take an additional `Arc>` and an `Arc<[PathBuf]>` workspace-roots snapshot, alongside the existing translator and subscriptions. `Translator::handle_cached_diagnostics` was removed and replaced by `Translator::cached_diagnostics_uri(workspace_roots, file_path) -> Result` and `Translator::diagnostics_from_cache_entry(Option<&DiagnosticInfo>) -> DiagnosticsResult`, so callers control exactly how long the `NotificationCache` lock is held between the two; `Translator::handle_server_logs`/`handle_server_messages` became associated functions taking `&NotificationCache` directly, since neither needed translator state. Acceptable pre-1.0. (#104) ### Fixed diff --git a/crates/mcpls-core/src/bridge/translator.rs b/crates/mcpls-core/src/bridge/translator.rs index 76e34e1..8a0e48a 100644 --- a/crates/mcpls-core/src/bridge/translator.rs +++ b/crates/mcpls-core/src/bridge/translator.rs @@ -19,7 +19,7 @@ use serde::{Deserialize, Serialize}; use tokio::time::Duration; use super::state::{ResourceLimits, detect_language, path_to_uri}; -use super::{DocumentTracker, NotificationCache}; +use super::{DiagnosticInfo, DocumentTracker, NotificationCache}; use crate::bridge::encoding::mcp_to_lsp_position; use crate::error::{Error, Result}; use crate::lsp::{LspClient, LspServer}; @@ -1472,58 +1472,60 @@ impl Translator { Ok(OutgoingCallsResult { calls }) } - /// Handle cached diagnostics request. + /// Resolve the LSP-side cache key (URI string) for a cached-diagnostics lookup. /// - /// Associated function rather than a method: it reads the caller-supplied - /// `workspace_roots` and `cache` instead of `self`, so callers that only - /// need a cached read (e.g. the `get_cached_diagnostics` MCP tool) never - /// need to lock `Arc>`, which may be held elsewhere - /// across a slow in-flight LSP round-trip. + /// Split out from the cache read itself so callers (e.g. the + /// `get_cached_diagnostics` MCP tool) can do the path `canonicalize()` and + /// workspace-boundary check *before* taking the `NotificationCache` lock — + /// that lock is also needed by `diagnostics_pump` to store incoming + /// notifications, so nothing that isn't a plain map lookup should run + /// while it's held. /// /// # Errors /// /// Returns an error if the path is invalid or outside workspace boundaries. - pub fn handle_cached_diagnostics( - workspace_roots: &[PathBuf], - file_path: &str, - cache: &NotificationCache, - ) -> Result { + pub fn cached_diagnostics_uri(workspace_roots: &[PathBuf], file_path: &str) -> Result { let path = PathBuf::from(file_path); let validated_path = validate_path_against_roots(&path, workspace_roots)?; // Use path_to_uri (strips \\?\ on Windows) so the key matches what // rust-analyzer stores in publishDiagnostics notifications. - let uri = path_to_uri(&validated_path).to_string(); - - let diagnostics = cache - .get_diagnostics(&uri) - .map_or_else(Vec::new, |diag_info| { - diag_info - .diagnostics - .iter() - .map(|diag| Diagnostic { - range: normalize_range(diag.range), - severity: match diag.severity { - Some(lsp_types::DiagnosticSeverity::ERROR) => DiagnosticSeverity::Error, - Some(lsp_types::DiagnosticSeverity::WARNING) => { - DiagnosticSeverity::Warning - } - Some(lsp_types::DiagnosticSeverity::INFORMATION) => { - DiagnosticSeverity::Information - } - Some(lsp_types::DiagnosticSeverity::HINT) => DiagnosticSeverity::Hint, - _ => DiagnosticSeverity::Information, - }, - message: diag.message.clone(), - code: diag.code.as_ref().map(|c| match c { - lsp_types::NumberOrString::Number(n) => n.to_string(), - lsp_types::NumberOrString::String(s) => s.clone(), - }), - }) - .collect() - }); + Ok(path_to_uri(&validated_path).to_string()) + } - Ok(DiagnosticsResult { diagnostics }) + /// Convert a cached diagnostics entry into the MCP-facing result shape. + /// + /// Takes an already-cloned `Option<&DiagnosticInfo>` (out of the + /// `NotificationCache` lock) rather than the cache itself, so this + /// mapping — which is not a bounded operation for a large diagnostics set + /// — never runs while the cache is locked. + #[must_use] + pub fn diagnostics_from_cache_entry(diag_info: Option<&DiagnosticInfo>) -> DiagnosticsResult { + let diagnostics = diag_info.map_or_else(Vec::new, |diag_info| { + diag_info + .diagnostics + .iter() + .map(|diag| Diagnostic { + range: normalize_range(diag.range), + severity: match diag.severity { + Some(lsp_types::DiagnosticSeverity::ERROR) => DiagnosticSeverity::Error, + Some(lsp_types::DiagnosticSeverity::WARNING) => DiagnosticSeverity::Warning, + Some(lsp_types::DiagnosticSeverity::INFORMATION) => { + DiagnosticSeverity::Information + } + Some(lsp_types::DiagnosticSeverity::HINT) => DiagnosticSeverity::Hint, + _ => DiagnosticSeverity::Information, + }, + message: diag.message.clone(), + code: diag.code.as_ref().map(|c| match c { + lsp_types::NumberOrString::Number(n) => n.to_string(), + lsp_types::NumberOrString::String(s) => s.clone(), + }), + }) + .collect() + }); + + DiagnosticsResult { diagnostics } } /// Handle server logs request. @@ -2741,10 +2743,10 @@ mod tests { let test_file = temp_dir.path().join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); - let result = - Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); - assert!(result.is_ok()); - let diags = result.unwrap(); + let cache_key = + Translator::cached_diagnostics_uri(&[], test_file.to_str().unwrap()).unwrap(); + let diag_info = cache.get_diagnostics(&cache_key).cloned(); + let diags = Translator::diagnostics_from_cache_entry(diag_info.as_ref()); assert_eq!(diags.diagnostics.len(), 0); } @@ -2852,10 +2854,10 @@ mod tests { cache.store_diagnostics(&uri, Some(1), vec![diagnostic]); - let result = - Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); - assert!(result.is_ok()); - let diags = result.unwrap(); + let cache_key = + Translator::cached_diagnostics_uri(&[], test_file.to_str().unwrap()).unwrap(); + let diag_info = cache.get_diagnostics(&cache_key).cloned(); + let diags = Translator::diagnostics_from_cache_entry(diag_info.as_ref()); assert_eq!(diags.diagnostics.len(), 1); assert_eq!(diags.diagnostics[0].message, "test error"); assert_eq!(diags.diagnostics[0].code, Some("E001".to_string())); @@ -2966,10 +2968,10 @@ mod tests { cache.store_diagnostics(&uri, Some(1), diagnostics); - let result = - Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); - assert!(result.is_ok()); - let diags = result.unwrap(); + let cache_key = + Translator::cached_diagnostics_uri(&[], test_file.to_str().unwrap()).unwrap(); + let diag_info = cache.get_diagnostics(&cache_key).cloned(); + let diags = Translator::diagnostics_from_cache_entry(diag_info.as_ref()); assert_eq!(diags.diagnostics.len(), 4); assert!(matches!( diags.diagnostics[0].severity, @@ -3025,19 +3027,17 @@ mod tests { cache.store_diagnostics(&uri, Some(1), vec![diagnostic]); - let result = - Translator::handle_cached_diagnostics(&[], test_file.to_str().unwrap(), &cache); - assert!(result.is_ok()); - let diags = result.unwrap(); + let cache_key = + Translator::cached_diagnostics_uri(&[], test_file.to_str().unwrap()).unwrap(); + let diag_info = cache.get_diagnostics(&cache_key).cloned(); + let diags = Translator::diagnostics_from_cache_entry(diag_info.as_ref()); assert_eq!(diags.diagnostics.len(), 1); assert_eq!(diags.diagnostics[0].code, Some("42".to_string())); } #[test] fn test_handle_cached_diagnostics_invalid_path() { - let cache = NotificationCache::new(); - let result = - Translator::handle_cached_diagnostics(&[], "/nonexistent/path/file.rs", &cache); + let result = Translator::cached_diagnostics_uri(&[], "/nonexistent/path/file.rs"); assert!(matches!(result, Err(Error::FileIo { .. }))); } @@ -3207,7 +3207,6 @@ mod tests { #[test] fn test_handle_cached_diagnostics_path_outside_workspace() { - let cache = NotificationCache::new(); let temp_dir1 = TempDir::new().unwrap(); let temp_dir2 = TempDir::new().unwrap(); @@ -3216,11 +3215,8 @@ mod tests { let test_file = temp_dir2.path().join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); - let result = Translator::handle_cached_diagnostics( - &workspace_roots, - test_file.to_str().unwrap(), - &cache, - ); + let result = + Translator::cached_diagnostics_uri(&workspace_roots, test_file.to_str().unwrap()); assert!(matches!(result, Err(Error::PathOutsideWorkspace(_)))); } diff --git a/crates/mcpls-core/src/mcp/server.rs b/crates/mcpls-core/src/mcp/server.rs index 266da9d..b6ad71c 100644 --- a/crates/mcpls-core/src/mcp/server.rs +++ b/crates/mcpls-core/src/mcp/server.rs @@ -387,10 +387,20 @@ impl McplsServer { &self, Parameters(CachedDiagnosticsParams { file_path }): Parameters, ) -> Result { - let result = { - let cache = self.context.notification_cache.lock().await; - Translator::handle_cached_diagnostics(&self.context.workspace_roots, &file_path, &cache) - }; + let result = + match Translator::cached_diagnostics_uri(&self.context.workspace_roots, &file_path) { + Ok(uri) => { + // Lock only long enough for the map lookup + clone: no + // canonicalize() or Vec mapping while `notification_cache` + // is held, since `diagnostics_pump` needs the same lock. + let diag_info = { + let cache = self.context.notification_cache.lock().await; + cache.get_diagnostics(&uri).cloned() + }; + Ok(Translator::diagnostics_from_cache_entry(diag_info.as_ref())) + } + Err(e) => Err(e), + }; match result { Ok(value) => serde_json::to_string(&value) @@ -604,10 +614,13 @@ impl ServerHandler for McplsServer { // Validated against a lock-free snapshot of workspace_roots (fixed at // startup) so this cache-only read never waits on the translator lock, // which may be held elsewhere across a slow in-flight LSP round-trip. - validate_path_against_roots(&path, &self.context.workspace_roots) + let validated_path = validate_path_against_roots(&path, &self.context.workspace_roots) .map_err(|e| McpError::invalid_params(e.to_string(), None))?; - let lsp_uri = crate::bridge::path_to_uri(&path); + // Build the URI from the canonicalized path (not the raw input path): + // it must match what `diagnostics_pump` stores from LSP notifications, + // which are always keyed by the canonical form. + let lsp_uri = crate::bridge::path_to_uri(&validated_path); // TODO(critic-S2): distinguish "file not tracked" from "file tracked but clean" // in the response shape. Currently both return `{"diagnostics":null}` which is @@ -928,6 +941,74 @@ mod tests { assert!(parsed.get("diagnostics").is_some()); } + /// `get_cached_diagnostics` end-to-end: a cache entry stored under the + /// canonical URI (as `diagnostics_pump` would store it) must be found when + /// requested via a textually non-canonical path -- proving `cached_diagnostics_uri` + /// still canonicalizes correctly after the lock-scope split, and that + /// `diagnostics_from_cache_entry` correctly maps a populated entry through + /// the actual tool call (not just the unit-level helpers directly). + #[tokio::test] + async fn test_cached_diagnostics_tool_finds_entry_via_noncanonical_path() { + use std::fs; + + use tempfile::TempDir; + use url::Url; + + let server = create_test_server(); + + let temp_dir = TempDir::new().unwrap(); + let subdir = temp_dir.path().join("sub"); + fs::create_dir(&subdir).unwrap(); + let test_file = subdir.join("test.rs"); + fs::write(&test_file, "fn main() {}").unwrap(); + + let canonical_path = test_file.canonicalize().unwrap(); + let uri: lsp_types::Uri = Url::from_file_path(&canonical_path) + .unwrap() + .as_str() + .parse() + .unwrap(); + let diagnostic = lsp_types::Diagnostic { + range: lsp_types::Range { + start: lsp_types::Position { + line: 0, + character: 0, + }, + end: lsp_types::Position { + line: 0, + character: 1, + }, + }, + severity: Some(lsp_types::DiagnosticSeverity::ERROR), + code: None, + code_description: None, + source: None, + message: "cached error".to_string(), + related_information: None, + tags: None, + data: None, + }; + { + let mut cache = server.context.notification_cache.lock().await; + cache.store_diagnostics(&uri, Some(1), vec![diagnostic]); + } + + // Textually distinct from `test_file`, but canonicalizes to the same path. + let noncanonical = subdir.join("..").join("sub").join("test.rs"); + let params = Parameters(CachedDiagnosticsParams { + file_path: noncanonical.to_str().unwrap().to_string(), + }); + + let result = server.get_cached_diagnostics(params).await; + assert!(result.is_ok()); + + let json_str = result.unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&json_str).unwrap(); + let diagnostics = parsed.get("diagnostics").unwrap().as_array().unwrap(); + assert_eq!(diagnostics.len(), 1); + assert_eq!(diagnostics[0].get("message").unwrap(), "cached error"); + } + #[tokio::test] async fn test_cached_diagnostics_tool_nonexistent_file() { let server = create_test_server(); @@ -1171,6 +1252,39 @@ mod tests { assert!(result.is_err()); } + /// Regression test for `read_resource`'s canonical-path fix: a path with + /// non-canonical segments (`..`) must resolve, via `validate_path_against_roots`, + /// to the same URI as its canonical form -- matching what `diagnostics_pump` + /// stores from LSP notifications. Building `lsp_uri` from the raw path (the + /// pre-fix behavior) would produce a mismatched cache key and always miss. + #[test] + fn test_read_resource_canonical_path_matches_pump_cache_key() { + use std::fs; + + use tempfile::TempDir; + + let temp_dir = TempDir::new().unwrap(); + let subdir = temp_dir.path().join("sub"); + fs::create_dir(&subdir).unwrap(); + let test_file = subdir.join("test.rs"); + fs::write(&test_file, "fn main() {}").unwrap(); + + // Textually distinct from `test_file`, but canonicalizes to the same path. + let noncanonical = subdir.join("..").join("sub").join("test.rs"); + assert_ne!(noncanonical, test_file); + + let validated = validate_path_against_roots(&noncanonical, &[]).unwrap(); + assert_eq!(validated, test_file.canonicalize().unwrap()); + + let uri_from_raw_path = crate::bridge::path_to_uri(&noncanonical); + let uri_from_validated_path = crate::bridge::path_to_uri(&validated); + assert_ne!( + uri_from_raw_path, uri_from_validated_path, + "raw and canonical paths must differ here, otherwise this test can't \ + detect a regression back to keying off the raw path" + ); + } + /// `validate_path` rejects a non-existent path (canonicalize fails). #[tokio::test] async fn test_validate_path_rejects_nonexistent_path() { From ba395a6dd436aeba9916f1590ed229f88507cbd9 Mon Sep 17 00:00:00 2001 From: "Andrei G." Date: Sat, 25 Jul 2026 00:59:40 +0200 Subject: [PATCH 3/3] fix(test): make canonical-path regression test symlink-based, not `..`-based `path_to_uri` re-parses the file URI through `url::Url::parse` for RFC 3986 character encoding, which normalizes away `..` segments on every platform. A path differing from its canonical form only by literal `..` segments therefore produces the same URI with or without the read_resource fix, regardless of OS. The test only passed on macOS because `/tmp` there is itself a symlink to `/private/tmp`, which `canonicalize()` resolves -- an effect unrelated to what the test claimed to prove. On Linux CI, `/tmp` is a real directory, so the assertion failed. Rewrite the test using a real symlinked directory so canonicalize() resolves something URI string normalization cannot, producing a deterministic raw-vs-canonical difference on any platform where symlinks work. Unix-only, since symlink creation on Windows CI runners typically requires elevated privileges. --- crates/mcpls-core/src/mcp/server.rs | 38 +++++++++++++++++++++-------- 1 file changed, 28 insertions(+), 10 deletions(-) diff --git a/crates/mcpls-core/src/mcp/server.rs b/crates/mcpls-core/src/mcp/server.rs index b6ad71c..b5349d9 100644 --- a/crates/mcpls-core/src/mcp/server.rs +++ b/crates/mcpls-core/src/mcp/server.rs @@ -1252,25 +1252,43 @@ mod tests { assert!(result.is_err()); } - /// Regression test for `read_resource`'s canonical-path fix: a path with - /// non-canonical segments (`..`) must resolve, via `validate_path_against_roots`, - /// to the same URI as its canonical form -- matching what `diagnostics_pump` - /// stores from LSP notifications. Building `lsp_uri` from the raw path (the - /// pre-fix behavior) would produce a mismatched cache key and always miss. + /// Regression test for `read_resource`'s canonical-path fix: a path reached + /// through a symlink must resolve, via `validate_path_against_roots`, to the + /// same URI as its canonical (symlink-resolved) form -- matching what + /// `diagnostics_pump` stores from LSP notifications. Building `lsp_uri` from + /// the raw (symlinked) path (the pre-fix behavior) would produce a + /// mismatched cache key and always miss. + /// + /// Uses a real symlink rather than `..` segments: `path_to_uri` re-parses + /// the URI string through `url::Url::parse` (for RFC 3986 char encoding), + /// which normalizes away `..` segments regardless of platform -- so a path + /// differing only by `..` produces the same URI as its canonical form with + /// or without the fix. Only an actual symlink resolution (which happens in + /// `canonicalize()`, not in URI string normalization) creates a real + /// raw-vs-canonical difference. Unix-only: creating symlinks on Windows CI + /// runners typically requires elevated privileges / Developer Mode. #[test] + #[cfg(unix)] fn test_read_resource_canonical_path_matches_pump_cache_key() { use std::fs; + use std::os::unix::fs::symlink; use tempfile::TempDir; let temp_dir = TempDir::new().unwrap(); - let subdir = temp_dir.path().join("sub"); - fs::create_dir(&subdir).unwrap(); - let test_file = subdir.join("test.rs"); + // Canonicalize the base up front so any symlink-iness already present + // in the OS temp directory itself (e.g. macOS's `/tmp` -> `/private/tmp`) + // doesn't leak into the comparison -- the only symlink under test is + // `link_dir`. + let base = temp_dir.path().canonicalize().unwrap(); + let real_dir = base.join("real"); + fs::create_dir(&real_dir).unwrap(); + let test_file = real_dir.join("test.rs"); fs::write(&test_file, "fn main() {}").unwrap(); - // Textually distinct from `test_file`, but canonicalizes to the same path. - let noncanonical = subdir.join("..").join("sub").join("test.rs"); + let link_dir = base.join("link"); + symlink(&real_dir, &link_dir).unwrap(); + let noncanonical = link_dir.join("test.rs"); assert_ne!(noncanonical, test_file); let validated = validate_path_against_roots(&noncanonical, &[]).unwrap();