Production safe analyzer

crates/analyzer/src/lib.rs

@@ -418,7 +418,7 @@ impl LanguageServer for Backend {
             .request::<request::GotoDefinition>(lsp_types::GotoDefinitionParams {
                 text_document_position_params: lsp_types::TextDocumentPositionParams {
                     position: lsp_types::Position {
-                        line: line as u32 - 1,
+                        line: line as u32,
                         character: character as u32,
                     },
                     text_document: lsp_types::TextDocumentIdentifier { uri },
@@ -504,6 +504,19 @@ impl LanguageServer for Backend {
             }
         }))
     }
+
+    async fn completion_resolve(&self, params: CompletionItem) -> Result<CompletionItem> {
+        match self
+            .rust_analyzer
+            .lock()
+            .await
+            .request::<request::ResolveCompletionItem>(params.clone())
+            .await
+        {
+            Ok(o) => Ok(o),
+            _ => Ok(params),
+        }
+    }
 }
 
 #[tokio::main]
@@ -601,18 +614,13 @@ fn find_row_col(rope: &Rope, needle: &str) -> Option<(usize, usize)> {
 
     let col_idx = char_idx - line_start;
 
-    Some((line_idx + 1, col_idx + 1))
+    Some((line_idx, col_idx + 1))
 }
 
 fn simplify_item(mut item: CompletionItem) -> CompletionItem {
     item.text_edit = None;
     item.additional_text_edits = None;
     item.command = None;
-    item.data = None;
-
-    item.insert_text = Some(item.label.clone());
-
-    item.insert_text_format = Some(InsertTextFormat::PLAIN_TEXT);
 
     item
 }

crates/analyzer/src/rust_analyzer.rs

@@ -1,10 +1,11 @@
-use std::{collections::HashMap, path::PathBuf, process::Stdio, sync::Arc};
+use std::{collections::HashMap, path::PathBuf, process::Stdio, sync::Arc, time::Duration};
 
 use serde::{Deserialize, Serialize, de::DeserializeOwned};
 use serde_json::{Value, json};
 use tokio::{
     io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
     sync::{Mutex, oneshot},
+    time::timeout,
 };
 use tower_lsp::lsp_types::{
     self, ClientCapabilities, InitializeParams, InitializedParams, Url, WorkspaceFolder,
@@ -12,10 +13,13 @@ use tower_lsp::lsp_types::{
     request::{self, Request},
 };
 
+const MAX_CONTENT_LENGTH: usize = 50 * 1024 * 1024;
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
 #[derive(Debug, Deserialize)]
 pub struct JsonRpcResponse<T> {
     pub jsonrpc: String,
-    pub id: Option<usize>,
+    pub id: Option<serde_json::Value>,
     pub result: Option<T>,
     pub error: Option<JsonRpcError>,
 }
@@ -27,13 +31,16 @@ pub struct JsonRpcError {
     pub data: Option<serde_json::Value>,
 }
 
-type PendingMap = Arc<Mutex<HashMap<usize, oneshot::Sender<Value>>>>;
+type PendingMap = Arc<Mutex<HashMap<usize, oneshot::Sender<Result<Value, String>>>>>;
 
 #[derive(Debug)]
 pub struct RustAnalyzer {
-    stdin: tokio::process::ChildStdin,
+    // Wrapped in a Mutex to support safe concurrent sharing if the design expands
+    stdin: Arc<Mutex<tokio::process::ChildStdin>>,
     pending: PendingMap,
     id: usize,
+    // Keep child handler to explicitly manage child process lifecycle and prevent zombie processes
+    _child: tokio::process::Child,
 }
 
 async fn send_lsp_message<W: AsyncWriteExt + Unpin>(
@@ -49,27 +56,37 @@ async fn send_lsp_message<W: AsyncWriteExt + Unpin>(
 
 async fn read_lsp_message<R: AsyncBufReadExt + Unpin>(
     reader: &mut R,
-) -> Result<String, Box<dyn std::error::Error>> {
+) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
     let mut line = String::new();
     let mut content_length = 0;
 
-    // Read headers until we hit the empty separator line (\r\n)
+    // Guard against infinite header reading attacks/bugs (Max 100 headers)
-    loop {
+    for _ in 0..100 {
         line.clear();
-        reader.read_line(&mut line).await?;
+        let bytes_read = reader.read_line(&mut line).await?;
-        if line == "\r\n" || line.is_empty() {
+        if bytes_read == 0 || line == "\r\n" || line.is_empty() {
             break;
         }
-        if line.to_lowercase().starts_with("content-length:") {
+        if line.to_ascii_lowercase().starts_with("content-length:") {
-            content_length = line["content-length:".len()..].trim().parse::<usize>()?;
+            if let Some(val_str) = line.split(':').nth(1) {
+                content_length = val_str.trim().parse::<usize>()?;
+            }
         }
     }
 
+    // Explode early if payload size violates strict guard rails to prevent memory-exhaustion (OOM)
     if content_length == 0 {
-        return Err("Missing or invalid Content-Length header".into());
+        return Err("Missing, invalid, or zero Content-Length header".into());
+    }
+    if content_length > MAX_CONTENT_LENGTH {
+        return Err(format!(
+            "Content-Length {} exceeds maximum threshold",
+            content_length
+        )
+        .into());
     }
 
-    // Read the exact byte buffer payload
+    // Explicitly secure pre-allocation limit
     let mut buffer = vec![0u8; content_length];
     reader.read_exact(&mut buffer).await?;
 
@@ -77,19 +94,26 @@ async fn read_lsp_message<R: AsyncBufReadExt + Unpin>(
 }
 
 impl RustAnalyzer {
-    pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
+    pub fn new() -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
         let mut child = tokio::process::Command::new("rust-analyzer")
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
-            .stderr(Stdio::null()) // Ignore logs for simplicity
+            .stderr(Stdio::null())
             .spawn()?;
 
-        let stdin = child.stdin.take().unwrap();
+        let stdin = child
-        let stdout = child.stdout.take().unwrap();
+            .stdin
+            .take()
+            .ok_or("Failed to open child stdin pipe")?;
+        let stdout = child
+            .stdout
+            .take()
+            .ok_or("Failed to open child stdout pipe")?;
 
         let pending: PendingMap = Arc::new(Mutex::new(HashMap::new()));
         let pending_clone = pending.clone();
 
+        // Background supervisor task loop
         tokio::spawn(async move {
             let mut stdout = BufReader::new(stdout);
 
@@ -97,7 +121,12 @@ impl RustAnalyzer {
                 let raw = match read_lsp_message(&mut stdout).await {
                     Ok(v) => v,
                     Err(err) => {
-                        eprintln!("LSP read error: {err}");
+                        eprintln!("LSP fatal stream read failure: {err}");
+                        // CRITICAL: Notify all pending channels that the bridge broke down
+                        let mut lock = pending_clone.lock().await;
+                        for (_, tx) in lock.drain() {
+                            let _ = tx.send(Err(format!("LSP reader task dropped: {}", err)));
+                        }
                         break;
                     }
                 };
@@ -105,51 +134,43 @@ impl RustAnalyzer {
                 let value: Value = match serde_json::from_str(&raw) {
                     Ok(v) => v,
                     Err(err) => {
-                        eprintln!("Invalid JSON from rust-analyzer: {err}");
+                        eprintln!("Corrupted JSON received: {err}");
-                        continue;
+                        continue; // Keep the connection running despite malformed frame
                     }
                 };
 
-                if value.get("method").is_some() {
+                // Filter server notification frames
+                if value.get("method").is_some() && value.get("id").is_none() {
                     continue;
                 }
 
                 let id = value.get("id").and_then(|v| v.as_u64()).map(|v| v as usize);
 
-                match id {
+                if let Some(id) = id {
-                    Some(id) => {
+                    let tx = pending_clone.lock().await.remove(&id);
-                        let tx = pending_clone.lock().await.remove(&id);
+                    if let Some(tx) = tx {
-
+                        let _ = tx.send(Ok(value));
-                        match tx {
+                    } else {
-                            Some(tx) => {
+                        eprintln!("Received orphaned or delayed frame for ID: {id}");
-                                let _ = tx.send(value);
-                            }
-                            None => {
-                                eprintln!("({:?}) {}", pending_clone, value);
-                                eprintln!("Received response for unknown request id {id}");
-                            }
-                        }
-                    }
-
-                    None => {
-                        // notification or server request
-                        eprintln!("Received server notification/request: {raw}");
                     }
+                } else {
+                    eprintln!("Received unhandled protocol notification framework: {raw}");
                 }
             }
         });
 
         Ok(Self {
-            stdin,
+            stdin: Arc::new(Mutex::new(stdin)),
             pending,
             id: 0,
+            _child: child,
         })
     }
 
     pub async fn request<R: Request>(
         &mut self,
         params: R::Params,
-    ) -> Result<R::Result, Box<dyn std::error::Error>>
+    ) -> Result<R::Result, Box<dyn std::error::Error + Send + Sync>>
     where
         R::Result: DeserializeOwned,
     {
@@ -160,47 +181,86 @@ impl RustAnalyzer {
 
         let (tx, rx) = oneshot::channel();
 
-        self.pending.lock().await.insert(id, tx);
+        // Scope the lock allocation tightly
-
+        {
-        send_lsp_message(
+            self.pending.lock().await.insert(id, tx);
-            &mut self.stdin,
-            &json!({
-                "jsonrpc": "2.0",
-                "id": id,
-                "method": R::METHOD,
-                "params": params,
-            }),
-        )
-        .await?;
-
-        let value = rx.await?;
-
-        let envelope: JsonRpcResponse<R::Result> = serde_json::from_value(value.clone())?;
-
-        if let Some(err) = envelope.error {
-            return Err(format!("LSP Error ({}): {}", err.code, err.message).into());
         }
 
-        envelope
+        let payload = json!({
-            .result
+            "jsonrpc": "2.0",
-            .ok_or_else(|| format!("missing result: {:?}", value).into())
+            "id": id,
+            "method": R::METHOD,
+            "params": params,
+        });
+
+        // Acquire lock on writing stream to ensure thread safety
+        let mut stdin_lock = self.stdin.lock().await;
+
+        if let Err(err) = send_lsp_message(&mut *stdin_lock, &payload).await {
+            // Rollback the pending map operation to avoid internal memory memory-leaks if serialization/IO errors trigger
+            self.pending.lock().await.remove(&id);
+            return Err(Box::new(err));
+        }
+
+        // Explicit drop of write lock early so other operations can pipe messages synchronously
+        drop(stdin_lock);
+
+        // Enforce an absolute time constraint limit to break free from hanging processes
+        let response_payload = match timeout(REQUEST_TIMEOUT, rx).await {
+            Ok(Ok(Ok(value))) => value,
+            Ok(Ok(Err(task_err))) => return Err(task_err.into()),
+            Ok(Err(_oneshot_canceled)) => {
+                return Err(
+                    "Bridge connection closed down; reader channel dropped unexpectedly".into(),
+                );
+            }
+            Err(_timeout_elapsed) => {
+                // Clear state tracking entries dynamically upon expiration failure
+                self.pending.lock().await.remove(&id);
+                return Err(
+                    format!("Request ID {} timed out after {:?}", id, REQUEST_TIMEOUT).into(),
+                );
+            }
+        };
+
+        let envelope: JsonRpcResponse<R::Result> =
+            serde_json::from_value(response_payload.clone())?;
+
+        if let Some(err) = envelope.error {
+            return Err(format!("LSP Engine Error ({}): {}", err.code, err.message).into());
+        }
+
+        envelope.result.ok_or_else(|| {
+            format!(
+                "Missing inner structural payload result: {:?}",
+                response_payload
+            )
+            .into()
+        })
     }
 
-    pub async fn notify<T: Serialize>(&mut self, method: &str, req: T) -> std::io::Result<()> {
+    pub async fn notify<T: Serialize>(
-        send_lsp_message(
+        &mut self,
-            &mut self.stdin,
+        method: &str,
-            &json!({
+        req: T,
-                "jsonrpc": "2.0",
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-                "method": method,
+        let payload = json!({
-                "params": req,
+            "jsonrpc": "2.0",
-            }),
+            "method": method,
-        )
+            "params": req,
-        .await
+        });
+
+        let mut stdin_lock = self.stdin.lock().await;
+        send_lsp_message(&mut *stdin_lock, &payload).await?;
+        Ok(())
     }
 }
 
 impl RustAnalyzer {
-    pub async fn initialize(&mut self, root: &PathBuf) -> Result<(), Box<dyn std::error::Error>> {
+    pub async fn initialize(
+        &mut self,
+        root: &PathBuf,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         let project_uri = Url::from_directory_path(root)
             .map_err(|_| "Failed to convert path to valid file:// URL")?;
 
@@ -226,14 +286,12 @@ impl RustAnalyzer {
         };
 
         self.request::<request::Initialize>(init_params).await?;
-
         Ok(())
     }
 
-    pub async fn initialized(&mut self) -> Result<(), Box<dyn std::error::Error>> {
+    pub async fn initialized(&mut self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         self.notify(Initialized::METHOD, InitializedParams {})
             .await?;
-
         Ok(())
     }
 }