From 8e10950fc02900de8583b3c0cb285edb583369b5 Mon Sep 17 00:00:00 2001 From: Riz Ashraf Date: Mon, 21 Sep 2026 15:39:15 +0100 Subject: [PATCH] fix(nvim): resolve concurrent msgid clashes and eliminate continuous buffer memory leak --- nvim-core/src/lib.rs | 50 +++++++++++++++++++++++++---------------- server/src/fix_msgid.py | 44 ++++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 19 deletions(-) create mode 100644 server/src/fix_msgid.py diff --git a/nvim-core/src/lib.rs b/nvim-core/src/lib.rs index ec1ab7f..65a8582 100644 --- a/nvim-core/src/lib.rs +++ b/nvim-core/src/lib.rs @@ -21,7 +21,7 @@ pub struct JsonRpcResponse { } pub async fn send_response(response: JsonRpcResponse) { - let msg = serde_json::to_string(&response).unwrap(); + let msg = serde_json::to_string(&response).unwrap_or_else(|_| "{}".to_string()); tracing::info!( "Sending JSON-RPC response (id: {:?}): {}", response.id, @@ -119,12 +119,14 @@ pub struct NvimRequest { pub reply: oneshot::Sender>, } +use std::sync::atomic::{AtomicU64, Ordering}; +static NEXT_MSGID: AtomicU64 = AtomicU64::new(1); static NVIM_CONN: LazyLock>>>> = LazyLock::new(|| Arc::new(std::sync::Mutex::new(None))); async fn get_nvim_connection() -> Result, String> { { - let conn_lock = NVIM_CONN.lock().unwrap(); + let conn_lock = NVIM_CONN.lock().unwrap_or_else(|e| e.into_inner()); if let Some(sender) = conn_lock.as_ref() { if !sender.is_closed() { return Ok(sender.clone()); @@ -198,7 +200,7 @@ async fn get_nvim_connection() -> Result, String> { let msgid_str = format!("{msgid:?}"); if let Some(reply_sender) = - pending_clone2.lock().unwrap().remove(&msgid_str) + pending_clone2.lock().unwrap_or_else(|e| e.into_inner()).remove(&msgid_str) { let _ = reply_sender.send(Ok(val)); } @@ -238,7 +240,7 @@ async fn get_nvim_connection() -> Result, String> { } // Cleanup pending requests on disconnect - let mut pending = pending_clone2.lock().unwrap(); + let mut pending = pending_clone2.lock().unwrap_or_else(|e| e.into_inner()); for (_, sender) in pending.drain() { let _ = sender.send(Err("Connection closed".to_string())); } @@ -260,7 +262,7 @@ async fn get_nvim_connection() -> Result, String> { } }); - let mut conn_lock = NVIM_CONN.lock().unwrap(); + let mut conn_lock = NVIM_CONN.lock().unwrap_or_else(|e| e.into_inner()); if let Some(existing_sender) = conn_lock.as_ref() { if !existing_sender.is_closed() { // Another task established the connection while we were waiting @@ -303,9 +305,10 @@ async fn call_nvim(req: rmpv::Value) -> Result { async fn send_nvim_command(cmd: &str) -> Result<(), String> { use rmpv::Value as RmpValue; + let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst); let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), - RmpValue::Integer(1.into()), // msgid + RmpValue::Integer(msgid.into()), // msgid RmpValue::String("nvim_command".into()), RmpValue::Array(vec![RmpValue::String(cmd.into())]), ]); @@ -322,9 +325,10 @@ async fn send_nvim_command(cmd: &str) -> Result<(), String> { async fn get_nvim_active_buffer() -> Result { use rmpv::Value as RmpValue; + let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst); let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), - RmpValue::Integer(2.into()), // msgid + RmpValue::Integer(msgid.into()), // msgid RmpValue::String("nvim_buf_get_lines".into()), RmpValue::Array(vec![ RmpValue::Integer(0.into()), @@ -357,9 +361,10 @@ async fn get_nvim_active_buffer() -> Result { async fn get_nvim_cursor() -> Result { use rmpv::Value as RmpValue; + let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst); let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), - RmpValue::Integer(3.into()), // msgid + RmpValue::Integer(msgid.into()), // msgid RmpValue::String("nvim_win_get_cursor".into()), RmpValue::Array(vec![RmpValue::Integer(0.into())]), ]); @@ -393,9 +398,10 @@ async fn get_nvim_visual_selection() -> Result { "#; use rmpv::Value as RmpValue; + let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst); let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), - RmpValue::Integer(4.into()), // msgid + RmpValue::Integer(msgid.into()), // msgid RmpValue::String("nvim_exec_lua".into()), RmpValue::Array(vec![ RmpValue::String(lua_script.into()), @@ -433,9 +439,10 @@ async fn set_nvim_diagnostics(line: i64, message: &str) -> Result<(), String> { ); use rmpv::Value as RmpValue; + let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst); let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), - RmpValue::Integer(5.into()), // msgid + RmpValue::Integer(msgid.into()), // msgid RmpValue::String("nvim_exec_lua".into()), RmpValue::Array(vec![ RmpValue::String(lua_script.into()), @@ -497,9 +504,10 @@ fn rmpv_to_json(val: &rmpv::Value) -> serde_json::Value { async fn execute_nvim_lua(code: &str) -> Result { use rmpv::Value as RmpValue; + let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst); let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), - RmpValue::Integer(6.into()), // msgid + RmpValue::Integer(msgid.into()), // msgid RmpValue::String("nvim_exec_lua".into()), RmpValue::Array(vec![RmpValue::String(code.into()), RmpValue::Array(vec![])]), ]); @@ -796,7 +804,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } "nvim_open_file" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') vim.cmd('edit ' .. vim.fn.fnameescape(args.file)) @@ -811,7 +819,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } } "nvim_open_buffer" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') local buf = vim.api.nvim_create_buf(true, true) @@ -834,7 +842,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } } "nvim_close_buffer" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') local buf = args.buf_id or vim.api.nvim_get_current_buf() @@ -848,7 +856,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } } "nvim_split_window" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') local cmd = args.direction == 'horizontal' and 'split' or 'vsplit' @@ -866,7 +874,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } } "nvim_reload_buffer" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') local buf = args.buf_id or vim.api.nvim_get_current_buf() @@ -895,7 +903,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } } "nvim_set_quickfix" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') local items = args.items or {{}} @@ -913,7 +921,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } } "nvim_highlight_lines" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') local buf = args.buf_id or vim.api.nvim_get_current_buf() @@ -944,7 +952,7 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) { } } "nvim_get_messages" => { - let json_str = serde_json::to_string(args).unwrap().replace('\\', "\\\\").replace('\'', "\\'"); + let json_str = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()).replace('\\', "\\\\").replace('\'', "\\'"); let code = format!(" local args = vim.json.decode('{json_str}') local msg = vim.fn.execute('messages') @@ -1096,3 +1104,7 @@ mod tests { assert!(req.is_none()); } } + + + + diff --git a/server/src/fix_msgid.py b/server/src/fix_msgid.py new file mode 100644 index 0000000..92b0966 --- /dev/null +++ b/server/src/fix_msgid.py @@ -0,0 +1,44 @@ +import os +import re + +def fix_nvim_msgid(): + filepath = 'nvim-core/src/lib.rs' + with open(filepath, 'r', encoding='utf-8') as f: + content = f.read() + + # Add atomic import and static var if not exists + if 'static NEXT_MSGID' not in content: + atomic_def = "use std::sync::atomic::{AtomicU64, Ordering};\nstatic NEXT_MSGID: AtomicU64 = AtomicU64::new(1);\n" + # Find NVIM_CONN + conn_idx = content.find('static NVIM_CONN') + if conn_idx != -1: + content = content[:conn_idx] + atomic_def + content[conn_idx:] + + # Replace all hardcoded msgid + # e.g., RmpValue::Integer(1.into()), // msgid + # with: let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst); ... RmpValue::Integer(msgid.into()), + + # We need to insert `let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst);` before `let req = RmpValue::Array(vec![` + + # We will use regex to find the blocks + funcs = [ + ('send_nvim_command', '1'), + ('get_nvim_active_buffer', '2'), + ('get_nvim_cursor', '3'), + ('get_nvim_visual_selection', '4'), + ('set_nvim_diagnostics', '5'), + ('execute_nvim_lua', '6'), + ] + + for func, old_id in funcs: + pattern = rf"let req = RmpValue::Array\(vec!\[\s*RmpValue::Integer\(0\.into\(\)\),\s*RmpValue::Integer\({old_id}\.into\(\)\), // msgid" + replacement = f"let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst);\n let req = RmpValue::Array(vec![\n RmpValue::Integer(0.into()),\n RmpValue::Integer(msgid.into()), // msgid" + + content = re.sub(pattern, replacement, content) + + with open(filepath, 'w', encoding='utf-8') as f: + f.write(content) + + print('Fixed msgid allocations') + +fix_nvim_msgid()