Fix memory leak in nvim-core by sweeping timed-out pending requests

nvim-core/src/lib.rs

@@ -203,11 +203,9 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
                     }
                     continue;
                 }
-                Err(_) => {
-                    if offset > 0 {
+                Err(rmpv::decode::Error::InvalidMarkerRead(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                     resp_buf.drain(..offset);
                     offset = 0;
-                    }
                     
                     let read_future = read_half.read(&mut chunk);
                     match tokio::time::timeout(tokio::time::Duration::from_secs(60), read_future).await {
@@ -220,6 +218,10 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
                         }
                     }
                 }
+                Err(_) => {
+                    tracing::error!("Failed to parse message from Neovim socket");
+                    break;
+                }
             }
         }
         
@@ -230,6 +232,16 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
         }
     });
 
+    // Cleanup task for timed-out requests
+    let pending_clone3 = Arc::clone(&pending_requests);
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(10));
+        loop {
+            interval.tick().await;
+            pending_clone3.lock().await.retain(|_, sender| !sender.is_closed());
+        }
+    });
+
     let mut conn_lock = NVIM_CONN.lock().await;
     *conn_lock = Some(tx.clone());
     Ok(tx)