From 3a78436665a886bffba023e6c4eb14e903e3681a Mon Sep 17 00:00:00 2001 From: Marco Arnulfo Date: Tue, 13 Oct 2026 05:40:08 +0200 Subject: [PATCH] fix(mcp): a slow bridge command no longer blocks every later call (part of #1538) --- crates/automation/src/bridge.rs | 114 ++++++++++++++++++++++++++------ 1 file changed, 95 insertions(+), 19 deletions(-) diff --git a/crates/automation/src/bridge.rs b/crates/automation/src/bridge.rs index 5fdc2f0d..f51db9ad 100644 --- a/crates/automation/src/bridge.rs +++ b/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::OwnedWriteHalf); pub struct BridgeClient { addr: String, - conn: Mutex>, + /// 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>, 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 { - 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 `", 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 = 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 { + 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 `", 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) -> (String, tokio::sync::mpsc::UnboundedReceiver) { + 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() {