Merge pull request #1548 from marcoarnulfo/fix/mcp-bridge-slow-command

fix(mcp): a slow bridge command no longer blocks every later call

crates/automation/src/bridge.rs

@@ -1,4 +1,5 @@
-//! Client for the desktop app's JSON-lines control protocol. One connection, reconnect on failure.
+//! Client for the desktop app's JSON-lines control protocol. A connection per request in flight,
+//! reused once its reply is in; reconnect on failure.
 
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
@@ -12,12 +13,17 @@ use crate::AutomationError;
 
 /// How often a blocking export's app job is polled.
 const JOB_POLL: Duration = Duration::from_millis(250);
+/// Idle connections kept for reuse; the app serves at most 16 at once.
+const MAX_IDLE: usize = 4;
 
 type Conn = (BufReader<tokio::net::tcp::OwnedReadHalf>, tokio::net::tcp::OwnedWriteHalf);
 
 pub struct BridgeClient {
     addr: String,
-    conn: Mutex<Option<Conn>>,
+    /// Connections with no request in flight. A request takes one (or opens one) for its whole
+    /// round trip and puts it back only once the reply is read, so a slow or abandoned request
+    /// never holds up the others or leaves its late reply for the next one to read (#1538).
+    idle: Mutex<Vec<Conn>>,
     next_id: AtomicU64,
 }
 
@@ -29,7 +35,7 @@ impl BridgeClient {
         if !matches!(host, "127.0.0.1" | "localhost" | "[::1]" | "::1") {
             return Err(AutomationError::BadRequest(format!("bridge address must be loopback, got `{addr}`")));
         }
-        Ok(Self { addr, conn: Mutex::new(None), next_id: AtomicU64::new(1) })
+        Ok(Self { addr, idle: Mutex::new(Vec::new()), next_id: AtomicU64::new(1) })
     }
 
     /// Run engine command `id` in the app. A blocking export (`file.exportMedia` with
@@ -71,20 +77,14 @@ impl BridgeClient {
 
     /// Call a control method; returns `result` or the app's error.
     pub async fn call(&self, method: &str, params: Value) -> Result<Value, AutomationError> {
-        let mut guard = self.conn.lock().await;
         for attempt in 0..2 {
-            if guard.is_none() {
-                let s = tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(&self.addr))
-                    .await
-                    .map_err(|_| AutomationError::Bridge(format!("timed out connecting to {}", self.addr)))?
-                    .map_err(|e| AutomationError::Bridge(format!("cannot connect to {} ({e}); start the app with `filmcraft --control <port>`", self.addr)))?;
-                let (r, w) = s.into_split();
-                *guard = Some((BufReader::new(r), w));
-            }
-            let id = self.next_id.fetch_add(1, Ordering::Relaxed);
-            let Some(conn) = guard.as_mut() else {
-                return Err(AutomationError::Bridge(format!("not connected to {}", self.addr)));
+            // A retry gets a fresh connection: the failed one may have been closed by an app restart.
+            let idle = if attempt == 0 { self.idle.lock().await.pop() } else { None };
+            let mut conn = match idle {
+                Some(c) => c,
+                None => self.connect().await?,
             };
+            let id = self.next_id.fetch_add(1, Ordering::Relaxed);
             let line = format!("{}\n", json!({"id": id, "method": method, "params": params}));
             let res: Result<Value, AutomationError> = async {
                 conn.1.write_all(line.as_bytes()).await.map_err(|e| AutomationError::Bridge(e.to_string()))?;
@@ -98,21 +98,32 @@ impl BridgeClient {
             .await;
             match res {
                 Ok(v) => {
+                    let mut idle = self.idle.lock().await;
+                    if idle.len() < MAX_IDLE {
+                        idle.push(conn);
+                    }
+                    drop(idle);
                     return if v.get("ok").and_then(Value::as_bool) == Some(true) {
                         Ok(v.get("result").cloned().unwrap_or(Value::Null))
                     } else {
                         Err(AutomationError::App(v.get("error").and_then(Value::as_str).unwrap_or("error").to_string()))
                     };
                 }
-                Err(e) if attempt == 0 => {
-                    *guard = None;
-                    let _ = e;
-                }
+                Err(_) if attempt == 0 => {}
                 Err(e) => return Err(e),
             }
         }
         Err(AutomationError::Bridge("unreachable".into()))
     }
+
+    async fn connect(&self) -> Result<Conn, AutomationError> {
+        let s = tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(&self.addr))
+            .await
+            .map_err(|_| AutomationError::Bridge(format!("timed out connecting to {}", self.addr)))?
+            .map_err(|e| AutomationError::Bridge(format!("cannot connect to {} ({e}); start the app with `filmcraft --control <port>`", self.addr)))?;
+        let (r, w) = s.into_split();
+        Ok((BufReader::new(r), w))
+    }
 }
 
 #[cfg(test)]
@@ -195,6 +206,71 @@ mod tests {
         assert!(e.to_string().contains("export failed: encode: disk full"), "{e}");
     }
 
+    /// A stand-in for the app that serves any number of connections and answers each request on
+    /// its own, as the app does: `slow` after `slow_reply` (never when `None`), anything else at
+    /// once with its command id. The ids of the requests it receives are sent to the returned channel.
+    async fn slow_app(slow_reply: Option<Duration>) -> (String, tokio::sync::mpsc::UnboundedReceiver<String>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap().to_string();
+        let (seen, rx) = tokio::sync::mpsc::unbounded_channel();
+        tokio::spawn(async move {
+            while let Ok((s, _)) = listener.accept().await {
+                let seen = seen.clone();
+                tokio::spawn(async move {
+                    let (r, w) = s.into_split();
+                    let w = Arc::new(Mutex::new(w));
+                    let mut lines = BufReader::new(r).lines();
+                    while let Ok(Some(line)) = lines.next_line().await {
+                        let req: Value = serde_json::from_str(&line).unwrap();
+                        let cmd = req["params"]["command"].as_str().unwrap_or_default().to_string();
+                        let _ = seen.send(cmd.clone());
+                        let delay = if cmd == "slow" { slow_reply } else { Some(Duration::ZERO) };
+                        let w = w.clone();
+                        tokio::spawn(async move {
+                            let Some(d) = delay else { return };
+                            tokio::time::sleep(d).await;
+                            let line = format!("{}\n", json!({"id": req["id"], "ok": true, "result": cmd}));
+                            let _ = w.lock().await.write_all(line.as_bytes()).await;
+                        });
+                    }
+                });
+            }
+        });
+        (addr, rx)
+    }
+
+    /// A command the app is slow to answer holds up only itself: a later call is answered at once
+    /// instead of queueing behind it, and behind every call queued after it (#1538).
+    #[tokio::test]
+    async fn a_slow_command_does_not_block_later_calls() {
+        let (addr, mut seen) = slow_app(None).await;
+        let b = Arc::new(BridgeClient::new(addr).unwrap());
+        let slow = tokio::spawn({
+            let b = b.clone();
+            async move { b.execute("slow", json!({})).await }
+        });
+        assert_eq!(seen.recv().await.as_deref(), Some("slow"));
+        let fast = tokio::time::timeout(Duration::from_secs(5), b.execute("fast", json!({}))).await;
+        assert_eq!(fast.expect("the later call waited on the slow one").unwrap(), json!("fast"));
+        slow.abort();
+    }
+
+    /// A call given up while it waits leaves no late reply on a connection for the next call to read
+    /// as its own (#1538).
+    #[tokio::test]
+    async fn an_abandoned_call_leaves_no_late_reply_behind() {
+        let (addr, mut seen) = slow_app(Some(Duration::from_millis(100))).await;
+        let b = Arc::new(BridgeClient::new(addr).unwrap());
+        let slow = tokio::spawn({
+            let b = b.clone();
+            async move { b.execute("slow", json!({})).await }
+        });
+        assert_eq!(seen.recv().await.as_deref(), Some("slow"));
+        slow.abort();
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        assert_eq!(b.execute("fast", json!({})).await.unwrap(), json!("fast"));
+    }
+
     /// Everything else is forwarded unchanged.
     #[tokio::test]
     async fn other_commands_are_forwarded() {