From 16366ce42202f74d58a62bafa4a5ac681a277349 Mon Sep 17 00:00:00 2001 From: KnottyDyes <47539659+KnottyDyes@users.noreply.github.com> Date: Mon, 12 Oct 2026 09:41:33 -0600 Subject: [PATCH] MCP: reconnect closed control sockets before sending requests After the app restarted, the remote backend wrote the next request to the old, closed connection and reported that the operation may have completed, although the restarted app received nothing. Reused connections are now checked without blocking before writing, and a closed one is replaced first. If the app can't be reached, the error says so and that nothing was sent. Failures after the request was written still report uncertainty and are never replayed. Loopback tests cover a restart between calls (the edit is applied exactly once) and a refused reconnect. The lost-reply test from #1246 still passes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BSB7fMQUccqFQekksDGWpo --- crates/mcp/src/backend.rs | 26 +++++++++- crates/mcp/src/tests.rs | 104 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+), 1 deletion(-) diff --git a/crates/mcp/src/backend.rs b/crates/mcp/src/backend.rs index 90bedd94..de30a899 100644 --- a/crates/mcp/src/backend.rs +++ b/crates/mcp/src/backend.rs @@ -42,6 +42,11 @@ impl Remote { &self.addr } + #[cfg(test)] + pub(crate) fn connection_probe(&self) -> TcpStream { + self.conn.as_ref().unwrap().1.try_clone().unwrap() + } + fn reconnect(&mut self) -> std::io::Result<()> { self.conn = None; let mut last = std::io::Error::new(std::io::ErrorKind::NotFound, format!("cannot resolve {}", self.addr)); @@ -61,10 +66,25 @@ impl Remote { Err(last) } - fn roundtrip(&mut self, line: &str) -> std::io::Result { + fn prepare_connection(&mut self) -> std::io::Result<()> { + if let Some((_, stream)) = self.conn.as_ref() { + stream.set_nonblocking(true)?; + let probe = stream.peek(&mut [0]); + stream.set_nonblocking(false)?; + match probe { + Ok(0) => self.conn = None, + Ok(_) => {} + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {} + Err(_) => self.conn = None, + } + } if self.conn.is_none() { self.reconnect()?; } + Ok(()) + } + + fn roundtrip(&mut self, line: &str) -> std::io::Result { let Some((reader, writer)) = self.conn.as_mut() else { return Err(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected to the app")); }; @@ -86,6 +106,10 @@ impl Remote { impl Backend for Remote { fn call(&mut self, method: &str, params: Value) -> Result { + self.prepare_connection().map_err(|e| { + self.conn = None; + format!("VectorCraft app at {} is not reachable: {e}; `{method}`: nothing was sent", self.addr) + })?; let id = self.next_id; self.next_id += 1; let line = json!({"id": id, "method": method, "params": params}).to_string(); diff --git a/crates/mcp/src/tests.rs b/crates/mcp/src/tests.rs index 603b486c..81fa137b 100644 --- a/crates/mcp/src/tests.rs +++ b/crates/mcp/src/tests.rs @@ -586,6 +586,110 @@ fn remote_connect_fails_fast() { drop(held); } +fn reusable_listener(addr: std::net::SocketAddr) -> std::net::TcpListener { + let socket = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None).unwrap(); + socket.set_reuse_address(true).unwrap(); + socket.bind(&addr.into()).unwrap(); + socket.listen(1).unwrap(); + socket.into() +} + +fn wait_for_remote_eof(remote: &Remote) { + let probe = remote.connection_probe(); + probe.set_read_timeout(Some(std::time::Duration::from_secs(2))).unwrap(); + assert_eq!(probe.peek(&mut [0]).unwrap(), 0); +} + +#[test] +fn remote_edit_succeeds_once_when_app_restarts() { + let listener = reusable_listener(std::net::SocketAddr::from(([127, 0, 0, 1], 0))); + let addr = listener.local_addr().unwrap(); + let mut remote = Remote::connect(&addr.to_string()).unwrap(); + let (old_connection, _) = listener.accept().unwrap(); + let old_handle = std::thread::spawn(move || { + let timeout = std::time::Duration::from_secs(2); + old_connection.set_read_timeout(Some(timeout)).unwrap(); + old_connection.set_write_timeout(Some(timeout)).unwrap(); + let mut reader = BufReader::new(old_connection.try_clone().unwrap()); + let mut writer = old_connection; + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + let message: Value = serde_json::from_str(&line).unwrap(); + assert_eq!(message["method"], "document.inspect"); + let result = Headless::with_document().call("document.inspect", message["params"].clone()).unwrap(); + writeln!(writer, "{}", json!({"id": message["id"], "ok": true, "result": result})).unwrap(); + writer + }); + let inspected = remote.call("document.inspect", json!({})).unwrap(); + assert!(inspected["objects"].as_u64().unwrap() > 0); + drop(old_handle.join().unwrap()); + drop(listener); + wait_for_remote_eof(&remote); + + let listener = reusable_listener(addr); + listener.set_nonblocking(true).unwrap(); + let handle = std::thread::spawn(move || { + let timeout = std::time::Duration::from_secs(2); + let deadline = std::time::Instant::now() + timeout; + let stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + if std::time::Instant::now() >= deadline { + return (0, 0); + } + std::thread::yield_now(); + } + Err(error) => panic!("accept failed: {error}"), + } + }; + stream.set_nonblocking(false).unwrap(); + stream.set_read_timeout(Some(timeout)).unwrap(); + stream.set_write_timeout(Some(timeout)).unwrap(); + let mut writer = stream.try_clone().unwrap(); + let mut backend = Headless::with_document(); + let base = backend.call("document.inspect", json!({})).unwrap()["objects"].as_u64().unwrap(); + let mut requests = 0; + for line in BufReader::new(stream).lines() { + let message: Value = serde_json::from_str(&line.unwrap()).unwrap(); + let result = backend.call(message["method"].as_str().unwrap(), message["params"].clone()).unwrap(); + requests += 1; + writeln!(writer, "{}", json!({"id": message["id"], "ok": true, "result": result})).unwrap(); + } + let objects = backend.call("document.inspect", json!({})).unwrap()["objects"].as_u64().unwrap(); + (requests, objects - base) + }); + + let edit = remote.call("engine.execute", json!({"command": "shape.rectangle", "params": {"x": 10, "y": 10, "width": 20, "height": 20}})); + drop(remote); + let (requests, edits) = handle.join().unwrap(); + + assert!(edit.is_ok(), "the first edit after restart must succeed: {edit:?}"); + assert_eq!((requests, edits), (1, 1), "the edit must be sent and applied exactly once"); +} + +#[test] +fn remote_call_reports_nothing_sent_when_reconnect_is_refused() { + let listener = reusable_listener(std::net::SocketAddr::from(([127, 0, 0, 1], 0))); + let addr = listener.local_addr().unwrap(); + let mut remote = Remote::connect(&addr.to_string()).unwrap(); + let (old_connection, _) = listener.accept().unwrap(); + drop(old_connection); + drop(listener); + wait_for_remote_eof(&remote); + let held = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None).unwrap(); + held.set_reuse_address(true).unwrap(); + held.bind(&addr.into()).unwrap(); + + let start = std::time::Instant::now(); + let error = remote.call("engine.execute", json!({"command": "shape.rectangle", "params": {"width": 20, "height": 20}})).unwrap_err(); + + assert!(start.elapsed() < std::time::Duration::from_secs(2)); + assert!(error.contains("not reachable"), "{error}"); + assert!(error.contains("nothing was sent"), "{error}"); + assert!(!error.contains("operation may have completed"), "{error}"); +} + #[test] fn headless_backend_methods() { let mut h = Headless::new();