Merge pull request #1739 from KnottyDyes/squad/remote-stale-connection

MCP: reconnect closed control sockets before sending requests

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,26 @@ impl Remote {
         Err(last)
     }
 
-    fn roundtrip(&mut self, line: &str) -> std::io::Result<String> {
+    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 {
+                // Closed by the app, or holding bytes nobody asked for (an app that explains why it
+                // closes): either way no reply to the next request can be read from it.
+                Ok(_) => self.conn = None,
+                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<String> {
         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 +107,10 @@ impl Remote {
 
 impl Backend for Remote {
     fn call(&mut self, method: &str, params: Value) -> Result<Value, String> {
+        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();

crates/mcp/src/tests.rs

@@ -586,6 +586,140 @@ 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");
+}
+
+/// A connection holding a line nobody asked for (an app that says why it closes an idle
+/// connection) is not used: the call goes out once, on a fresh connection, and succeeds.
+#[test]
+fn remote_does_not_send_on_a_connection_with_a_stray_line() {
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    let mut remote = Remote::connect(&addr).unwrap();
+    let (mut stale, _) = listener.accept().unwrap();
+    writeln!(stale, "{}", json!({"ok": false, "error": "idle"})).unwrap();
+    let probe = remote.connection_probe();
+    probe.set_read_timeout(Some(std::time::Duration::from_secs(2))).unwrap();
+    assert_eq!(probe.peek(&mut [0]).unwrap(), 1, "the stray line arrived");
+    drop(probe);
+    let h = std::thread::spawn(move || {
+        let (stream, _) = listener.accept().unwrap();
+        stream.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
+        let mut out = stream.try_clone().unwrap();
+        let mut line = String::new();
+        BufReader::new(stream).read_line(&mut line).unwrap();
+        let msg: Value = serde_json::from_str(&line).unwrap();
+        writeln!(out, "{}", json!({"ok": true, "result": 7, "id": msg["id"]})).unwrap();
+    });
+    assert_eq!(remote.call("document.inspect", json!({})), Ok(json!(7)));
+    h.join().unwrap();
+    // Nothing was written to the stale connection: it ends (or is reset, its line unread) without a byte.
+    stale.set_read_timeout(Some(std::time::Duration::from_secs(2))).unwrap();
+    let mut rest = String::new();
+    assert_eq!(BufReader::new(stale).read_line(&mut rest).unwrap_or(0), 0, "{rest}");
+}
+
+#[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();

docs/mcp.md

@@ -111,7 +111,11 @@ to) before it is dispatched. When a tool call reaches a running app, it is one c
 request, and the app echoes its `id` in the reply: a reply with any other `id` is rejected, the
 connection dropped, and the next call starts a fresh one — the call reports uncertainty like any
 reply that never arrived whole. The app's refusals carry no `id` (a connection past its 16, a line it
-would not read): the call reports the app's reason and says nothing was run. Clients requesting `2026-07-28` through `initialize`, or through a request's
+would not read): the call reports the app's reason and says nothing was run. Before each call the
+server checks that the app still holds the connection and opens a new one when it doesn't (the app
+closes a connection left idle for 5 minutes, and a restarted app is a new one), so a request is
+never written to a closed socket; when the app can't be reached, the call says that nothing was
+sent. Clients requesting `2026-07-28` through `initialize`, or through a request's
 `_meta["io.modelcontextprotocol/protocolVersion"]`, receive `resultType: "complete"`,
 `ttlMs` and `cacheScope: "private"` on tool/resource/template/prompt lists and resource/prompt reads.
 List TTL is 600000 ms; live reads use zero. Older clients keep the original response shapes.