From e8d7dc24f21ca3504ce4692c3fe739a5053dd7fb Mon Sep 17 00:00:00 2001 From: Euisuh Jeong <55032583+euisuh@users.noreply.github.com> Date: Wed, 14 Oct 2026 07:18:40 +0900 Subject: [PATCH] fix(control): bound frames and expire queued requests (#213) MCP and control-channel lines are bounded while they are read (32 MiB; an oversized frame gets an error and the stream closes without being drained). The loopback control listener caps connections at 8, applies 60 s socket read and write timeouts, caps both request queues at 8 and discards requests whose client stopped waiting. Fixes #201. --- apps/figcraft/src/control_server.rs | 86 +++++-- apps/figcraft/src/control_server_tests.rs | 262 ++++++++++++++++++++++ apps/figcraft/src/main.rs | 4 +- crates/mcp/src/server.rs | 28 ++- crates/mcp/tests/protocol.rs | 36 +++ crates/ui-egui/src/control.rs | 11 + crates/ui-egui/src/idle_tests.rs | 10 +- docs/control-protocol.md | 14 ++ docs/mcp.md | 5 + 9 files changed, 427 insertions(+), 29 deletions(-) create mode 100644 apps/figcraft/src/control_server_tests.rs diff --git a/apps/figcraft/src/control_server.rs b/apps/figcraft/src/control_server.rs index fbc3802..5cfe129 100644 --- a/apps/figcraft/src/control_server.rs +++ b/apps/figcraft/src/control_server.rs @@ -1,7 +1,7 @@ //! Localhost JSON-lines control server: each line `{"id", "method", "params"}` is queued for the UI //! thread; the reply line `{"id", "ok", "result" | "error"}` is written when the frame handled it. -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -12,6 +12,20 @@ use serde_json::{Value, json}; type Queue = Arc>>; type CtxSlot = Arc>>; +const MAX_LINE: usize = 32 * 1024 * 1024; +const MAX_CONNECTIONS: usize = 8; +const MAX_QUEUED_REQUESTS: usize = 8; +const TIMEOUT: Duration = Duration::from_secs(60); + +/// Deferred requests for a hidden window must not make the UI's queue grow without a bound. +pub fn drain_queue(queue: &Queue, target: &mut Vec) { + target.retain(|request| !request.is_cancelled()); + let mut queue = queue.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + queue.retain(|request| !request.is_cancelled()); + let count = queue.len().min(MAX_QUEUED_REQUESTS.saturating_sub(target.len())); + target.extend(queue.drain(..count)); +} + pub fn start(port: u16, queue: Queue, ctx: CtxSlot) { let listener = match TcpListener::bind(("127.0.0.1", port)) { Ok(l) => l, @@ -23,9 +37,9 @@ pub fn start(port: u16, queue: Queue, ctx: CtxSlot) { }; eprintln!("FigCraft control server listening on 127.0.0.1:{port}"); let spawned = std::thread::Builder::new().name("figcraft-control".into()).spawn(move || { + let mut workers = Vec::new(); for stream in listener.incoming().flatten() { - let (q, c) = (queue.clone(), ctx.clone()); - let _ = std::thread::Builder::new().name("figcraft-control-conn".into()).spawn(move || serve(stream, q, c)); + spawn_connection(stream, queue.clone(), ctx.clone(), &mut workers, TIMEOUT); } }); if let Err(e) = spawned { @@ -33,37 +47,77 @@ pub fn start(port: u16, queue: Queue, ctx: CtxSlot) { } } -fn serve(stream: TcpStream, queue: Queue, ctx: CtxSlot) { - let Ok(read_half) = stream.try_clone() else { return }; - let mut out = stream; - let reader = BufReader::new(read_half); - for line in reader.lines() { - let Ok(line) = line else { break }; +fn spawn_connection(stream: TcpStream, queue: Queue, ctx: CtxSlot, workers: &mut Vec>, timeout: Duration) { + workers.retain(|worker| !worker.is_finished()); + if workers.len() >= MAX_CONNECTIONS { + return; + } + match std::thread::Builder::new().name("figcraft-control-conn".into()).spawn(move || { + let _ = serve(stream, queue, ctx, timeout); + }) { + Ok(worker) => workers.push(worker), + Err(e) => log::error!("control connection thread: {e}"), + } +} + +fn serve(mut out: TcpStream, queue: Queue, ctx: CtxSlot, timeout: Duration) -> std::io::Result<()> { + out.set_read_timeout(Some(timeout))?; + out.set_write_timeout(Some(timeout))?; + let mut reader = BufReader::new(out.try_clone()?); + let mut line = Vec::new(); + loop { + line.clear(); + if (&mut reader).take((MAX_LINE + 1) as u64).read_until(b'\n', &mut line)? == 0 { + break; + } + if line.last() == Some(&b'\n') { + line.pop(); + } + if line.len() > MAX_LINE { + writeln!(out, "{}", json!({"id": null, "ok": false, "error": "message too large"}))?; + break; + } + let Ok(line) = std::str::from_utf8(&line) else { break }; if line.trim().is_empty() { continue; } - let reply = match serde_json::from_str::(&line) { + let reply = match serde_json::from_str::(line) { Ok(v) => { let id = v.get("id").cloned().unwrap_or(Value::Null); let method = v.get("method").and_then(Value::as_str).unwrap_or("").to_string(); let params = v.get("params").cloned().unwrap_or(json!({})); let (tx, rx) = std::sync::mpsc::channel(); - queue.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push(ControlRequest { id: id.clone(), method, params, reply: tx }); + let cancelled = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let mut pending = queue.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + pending.retain(|request| !request.is_cancelled()); + if pending.len() >= MAX_QUEUED_REQUESTS { + drop(pending); + writeln!(out, "{}", json!({"id": id, "ok": false, "error": "control request queue is full"}))?; + continue; + } + pending.push(ControlRequest { id: id.clone(), method, params, reply: tx, cancelled: cancelled.clone() }); + drop(pending); if let Some(c) = ctx.lock().unwrap_or_else(std::sync::PoisonError::into_inner).as_ref() { c.request_repaint(); } - match rx.recv_timeout(Duration::from_secs(60)) { + match rx.recv_timeout(timeout) { Ok(r) => match r.result { Ok(res) => json!({"id": r.id, "ok": true, "result": res}), Err(e) => json!({"id": r.id, "ok": false, "error": e}), }, - Err(_) => json!({"id": id, "ok": false, "error": "timed out waiting for the UI (is the window visible?)"}), + Err(_) => { + cancelled.store(true, std::sync::atomic::Ordering::Relaxed); + json!({"id": id, "ok": false, "error": "timed out waiting for the UI (is the window visible?)"}) + } } } Err(e) => json!({"id": null, "ok": false, "error": format!("bad JSON: {e}")}), }; - if writeln!(out, "{reply}").is_err() { - break; - } + writeln!(out, "{reply}")?; } + Ok(()) } + +#[cfg(test)] +#[path = "control_server_tests.rs"] +mod tests; diff --git a/apps/figcraft/src/control_server_tests.rs b/apps/figcraft/src/control_server_tests.rs new file mode 100644 index 0000000..2e0a7b4 --- /dev/null +++ b/apps/figcraft/src/control_server_tests.rs @@ -0,0 +1,262 @@ +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use super::*; +use std::io::Read; +use std::net::Shutdown; +use std::sync::mpsc; +use std::time::Instant; + +fn pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let client = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + client.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + client.set_write_timeout(Some(Duration::from_secs(5))).unwrap(); + (client, listener.accept().unwrap().0) +} + +fn queue() -> Queue { + Arc::new(Mutex::new(Vec::new())) +} + +fn context() -> CtxSlot { + Arc::new(Mutex::new(None)) +} + +fn reply(reader: &mut BufReader) -> Value { + let mut line = String::new(); + assert!(reader.read_line(&mut line).unwrap() > 0); + serde_json::from_str(&line).unwrap() +} + +fn request() -> ControlRequest { + ControlRequest { id: json!(0), method: "ui.inspect".into(), params: json!({}), reply: mpsc::channel().0, cancelled: Default::default() } +} + +#[test] +fn rejects_an_oversized_unterminated_frame_without_dispatch() { + let (mut client, server) = pair(); + let pending = queue(); + let q = pending.clone(); + let worker = std::thread::spawn(move || serve(server, q, context(), Duration::from_secs(5))); + // Generate bytes as they are written; no giant input string and no terminating newline. + std::io::copy(&mut std::io::repeat(b' ').take((MAX_LINE + 1) as u64), &mut client).unwrap(); + let mut reader = BufReader::new(client); + assert_eq!(reply(&mut reader), json!({"id": null, "ok": false, "error": "message too large"})); + assert_eq!(reader.read(&mut [0]).unwrap(), 0); + worker.join().unwrap().unwrap(); + assert!(pending.lock().unwrap().is_empty()); +} + +#[test] +fn valid_frames_keep_their_order_ids_and_parameters() { + let (mut client, server) = pair(); + let pending = queue(); + let q = pending.clone(); + let worker = std::thread::spawn(move || serve(server, q, context(), Duration::from_secs(5))); + client.write_all(b"\r\n{bad JSON}\n{\"id\":\"test\",\"method\":\"engine.execute\",\"params\":{\"command\":\"shape.rectangle\"}}\r\n").unwrap(); + let mut reader = BufReader::new(client); + assert_eq!(reply(&mut reader)["ok"], false); + let start = Instant::now(); + let r = loop { + if let Some(r) = pending.lock().unwrap().pop() { + break r; + } + assert!(start.elapsed() < Duration::from_secs(5)); + std::thread::sleep(Duration::from_millis(1)); + }; + assert_eq!(r.id, "test"); + assert_eq!(r.method, "engine.execute"); + assert_eq!(r.params, json!({"command": "shape.rectangle"})); + r.reply.send(figcraft_ui_egui::ControlReply { id: r.id, result: Ok(json!({"id": 42})) }).unwrap(); + assert_eq!(reply(&mut reader), json!({"id": "test", "ok": true, "result": {"id": 42}})); + reader.get_ref().shutdown(Shutdown::Write).unwrap(); + worker.join().unwrap().unwrap(); +} + +#[test] +fn full_queue_rejects_before_dispatch_and_ui_queue_stays_bounded() { + let (mut client, server) = pair(); + let pending = queue(); + pending.lock().unwrap().extend((0..MAX_QUEUED_REQUESTS).map(|_| request())); + let mut ui_queue = (0..MAX_QUEUED_REQUESTS - 1).map(|_| request()).collect(); + drain_queue(&pending, &mut ui_queue); + assert_eq!(ui_queue.len(), MAX_QUEUED_REQUESTS); + assert_eq!(pending.lock().unwrap().len(), MAX_QUEUED_REQUESTS - 1); + pending.lock().unwrap().push(request()); + drain_queue(&pending, &mut ui_queue); + assert_eq!(pending.lock().unwrap().len(), MAX_QUEUED_REQUESTS); + let q = pending.clone(); + let worker = std::thread::spawn(move || serve(server, q, context(), Duration::from_secs(5))); + client.write_all(b"{\"id\":1,\"method\":\"engine.execute\"}\n").unwrap(); + let mut reader = BufReader::new(client); + assert_eq!(reply(&mut reader), json!({"id": 1, "ok": false, "error": "control request queue is full"})); + assert_eq!(pending.lock().unwrap().len(), MAX_QUEUED_REQUESTS); + ui_queue.clear(); + drain_queue(&pending, &mut ui_queue); + assert_eq!(ui_queue.len(), MAX_QUEUED_REQUESTS); + assert!(pending.lock().unwrap().is_empty()); + reader.get_ref().shutdown(Shutdown::Write).unwrap(); + worker.join().unwrap().unwrap(); +} + +#[test] +fn idle_and_partial_frame_connections_expire() { + for bytes in [b"".as_slice(), b"{\"id\":"] { + let (mut client, server) = pair(); + let (done, finished) = mpsc::channel(); + let pending = queue(); + let q = pending.clone(); + let worker = std::thread::spawn(move || done.send(serve(server, q, context(), Duration::from_millis(50)))); + client.write_all(bytes).unwrap(); + let err = finished.recv_timeout(Duration::from_secs(5)).unwrap().unwrap_err(); + assert!(matches!(err.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut)); + assert!(pending.lock().unwrap().is_empty()); + worker.join().unwrap().unwrap(); + } +} + +#[test] +fn connection_limit_rejects_excess_clients_and_reuses_finished_slots() { + let pending = queue(); + let ctx = context(); + let mut workers = Vec::new(); + let mut clients = Vec::new(); + for _ in 0..MAX_CONNECTIONS { + let (client, server) = pair(); + spawn_connection(server, pending.clone(), ctx.clone(), &mut workers, Duration::from_secs(5)); + clients.push(client); + } + assert_eq!(workers.len(), MAX_CONNECTIONS); + let (mut extra, server) = pair(); + spawn_connection(server, pending.clone(), ctx.clone(), &mut workers, Duration::from_secs(5)); + assert_eq!(extra.read(&mut [0]).unwrap(), 0); + assert_eq!(workers.len(), MAX_CONNECTIONS); + for client in clients { + client.shutdown(Shutdown::Both).unwrap(); + } + let start = Instant::now(); + while workers.iter().any(|worker| !worker.is_finished()) { + assert!(start.elapsed() < Duration::from_secs(5)); + std::thread::sleep(Duration::from_millis(1)); + } + let (mut client, server) = pair(); + spawn_connection(server, pending, ctx, &mut workers, Duration::from_secs(5)); + assert_eq!(workers.len(), 1); + client.write_all(b"invalid\n").unwrap(); + assert_eq!(reply(&mut BufReader::new(client.try_clone().unwrap()))["ok"], false); + client.shutdown(Shutdown::Both).unwrap(); + workers.pop().unwrap().join().unwrap(); +} + +#[test] +fn a_client_that_stops_reading_cannot_hold_a_worker() { + let (mut client, server) = pair(); + let pending = queue(); + let q = pending.clone(); + let (done, finished) = mpsc::channel(); + let worker = std::thread::spawn(move || done.send(serve(server, q, context(), Duration::from_millis(100)))); + client.write_all(b"{\"id\":1,\"method\":\"document.inspect\"}\n").unwrap(); + let start = Instant::now(); + let r = loop { + if let Some(r) = pending.lock().unwrap().pop() { + break r; + } + assert!(start.elapsed() < Duration::from_secs(5)); + std::thread::sleep(Duration::from_millis(1)); + }; + r.reply.send(figcraft_ui_egui::ControlReply { id: r.id, result: Ok(json!("x".repeat(8 * 1024 * 1024))) }).unwrap(); + let err = finished.recv_timeout(Duration::from_secs(5)).unwrap().unwrap_err(); + assert!(matches!(err.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut)); + worker.join().unwrap().unwrap(); +} + +#[test] +fn a_limit_sized_frame_is_accepted_at_eof() { + let (mut client, server) = pair(); + let pending = queue(); + let q = pending.clone(); + let worker = std::thread::spawn(move || serve(server, q, context(), Duration::from_secs(5))); + let message = b"{\"id\":1,\"method\":\"ui.inspect\"}"; + client.write_all(message).unwrap(); + std::io::copy(&mut std::io::repeat(b' ').take((MAX_LINE - message.len()) as u64), &mut client).unwrap(); + client.shutdown(Shutdown::Write).unwrap(); + let start = Instant::now(); + let r = loop { + if let Some(r) = pending.lock().unwrap().pop() { + break r; + } + assert!(start.elapsed() < Duration::from_secs(5)); + std::thread::sleep(Duration::from_millis(1)); + }; + assert_eq!(r.id, 1); + assert_eq!(r.method, "ui.inspect"); + r.reply.send(figcraft_ui_egui::ControlReply { id: r.id, result: Ok(json!({})) }).unwrap(); + assert_eq!(reply(&mut BufReader::new(client)), json!({"id": 1, "ok": true, "result": {}})); + worker.join().unwrap().unwrap(); +} + +#[test] +fn timed_out_requests_release_a_full_deferred_ui_queue() { + use figcraft_ui_egui::{FigApp, NoServices, control, windows}; + + let ctx = egui::Context::default(); + ctx.set_embed_viewports(false); + let mut app = FigApp::new(figcraft_engine::Session::new(), Box::new(NoServices)); + let window = windows::open(&mut app, &ctx, Some(0), None, None).unwrap(); + let pending = queue(); + let mut clients = Vec::new(); + let mut workers = Vec::new(); + for id in 0..MAX_QUEUED_REQUESTS { + let (mut client, server) = pair(); + let q = pending.clone(); + workers.push(std::thread::spawn(move || serve(server, q, context(), Duration::from_secs(1)))); + writeln!(client, "{}", json!({"id": id, "method": "engine.execute", "params": {"window": window, "command": "file.new"}})).unwrap(); + clients.push(BufReader::new(client)); + } + let start = Instant::now(); + while pending.lock().unwrap().len() < MAX_QUEUED_REQUESTS { + assert!(start.elapsed() < Duration::from_secs(5)); + std::thread::sleep(Duration::from_millis(1)); + } + drain_queue(&pending, &mut app.control_queue); + control::drain(&mut app, &ctx); + assert_eq!(app.control_queue.len(), MAX_QUEUED_REQUESTS, "the other window has not drawn"); + for client in &mut clients { + assert!(reply(client)["error"].as_str().unwrap().contains("timed out")); + client.get_ref().shutdown(Shutdown::Write).unwrap(); + } + assert!(app.control_queue.iter().all(ControlRequest::is_cancelled)); + let expired = app.control_queue.pop().unwrap(); + // Include expired requests in the socket queue, too: admission must reclaim their slots. + pending.lock().unwrap().extend((0..MAX_QUEUED_REQUESTS).map(|_| { + let r = request(); + r.cancelled.store(true, std::sync::atomic::Ordering::Relaxed); + r + })); + let (mut client, server) = pair(); + let q = pending.clone(); + workers.push(std::thread::spawn(move || serve(server, q, context(), Duration::from_secs(5)))); + client.write_all(b"{\"id\":100,\"method\":\"window.list\",\"params\":{\"window\":0}}\n").unwrap(); + let start = Instant::now(); + while !pending.lock().unwrap().iter().any(|r| r.id == 100) { + assert!(start.elapsed() < Duration::from_secs(5)); + std::thread::sleep(Duration::from_millis(1)); + } + drain_queue(&pending, &mut app.control_queue); + assert_eq!(app.control_queue.len(), 1, "expired UI requests no longer occupy the cap"); + control::drain(&mut app, &ctx); + let mut client = BufReader::new(client); + assert_eq!(reply(&mut client)["ok"], true); + client.get_ref().shutdown(Shutdown::Write).unwrap(); + assert!(app.control_queue.is_empty()); + // A window that finally draws must discard its expired command before dispatch, too. + let documents = app.session.docs.len(); + app.windows.current = window; + app.control_queue.push(expired); + control::drain(&mut app, &ctx); + assert!(app.control_queue.is_empty()); + assert_eq!(app.session.docs.len(), documents); + for worker in workers { + worker.join().unwrap().unwrap(); + } +} diff --git a/apps/figcraft/src/main.rs b/apps/figcraft/src/main.rs index 8ef814c..7689fb7 100644 --- a/apps/figcraft/src/main.rs +++ b/apps/figcraft/src/main.rs @@ -51,9 +51,7 @@ impl eframe::App for App { } } let title = with_app(|fig| { - let mut q = self.queue.lock().unwrap_or_else(std::sync::PoisonError::into_inner); - fig.control_queue.extend(q.drain(..)); - drop(q); + control_server::drain_queue(&self.queue, &mut fig.control_queue); // Closing the window asks about unsaved changes (and keeps the app running while other // windows are open). figcraft_ui_egui::windows::on_close_request(fig, ctx); diff --git a/crates/mcp/src/server.rs b/crates/mcp/src/server.rs index 59acd2f..365c857 100644 --- a/crates/mcp/src/server.rs +++ b/crates/mcp/src/server.rs @@ -1,6 +1,6 @@ //! JSON-RPC 2.0 framing and the MCP lifecycle / tools / resources / prompts methods. -use std::io::{BufRead, Write}; +use std::io::{BufRead, Read, Write}; use serde_json::{Value, json}; @@ -76,10 +76,22 @@ impl Server { } /// Serve newline-delimited JSON-RPC until `input` closes. stdout is the protocol stream; logs - /// go to stderr. - pub fn serve(&mut self, input: impl BufRead, mut output: impl Write) -> std::io::Result<()> { - for line in input.split(b'\n') { - let line = line?; + /// go to stderr. An oversized frame gets an error and closes the stream without draining it. + pub fn serve(&mut self, mut input: impl BufRead, mut output: impl Write) -> std::io::Result<()> { + let mut line = Vec::new(); + loop { + line.clear(); + if (&mut input).take((MAX_LINE + 1) as u64).read_until(b'\n', &mut line)? == 0 { + break; + } + if line.last() == Some(&b'\n') { + line.pop(); + } + if line.len() > MAX_LINE { + writeln!(output, "{}", error(Value::Null, INVALID_REQUEST, "message too large"))?; + output.flush()?; + break; + } let text = String::from_utf8_lossy(&line); if let Some(reply) = self.handle_line(&text) { output.write_all(reply.as_bytes())?; @@ -92,13 +104,13 @@ impl Server { /// Handle one line; returns the reply line (None for notifications and blank lines). pub fn handle_line(&mut self, line: &str) -> Option { + if line.len() > MAX_LINE { + return Some(error(Value::Null, INVALID_REQUEST, "message too large").to_string()); + } let line = line.trim(); if line.is_empty() { return None; } - if line.len() > MAX_LINE { - return Some(error(Value::Null, INVALID_REQUEST, "message too large").to_string()); - } let reply = match serde_json::from_str::(line) { Ok(Value::Array(batch)) => { // Batches were removed in 2025-06-18; still answer older clients sensibly. diff --git a/crates/mcp/tests/protocol.rs b/crates/mcp/tests/protocol.rs index 6c9f761..89f2a64 100644 --- a/crates/mcp/tests/protocol.rs +++ b/crates/mcp/tests/protocol.rs @@ -172,3 +172,39 @@ fn serve_over_streams() { assert_eq!(lines[1]["error"]["code"], -32700); assert_eq!(lines[2]["id"], 2); } + +#[test] +fn serve_rejects_oversized_frames_before_consuming_the_stream() { + use std::io::{BufReader, Read}; + + const LIMIT: u64 = 32 * 1024 * 1024; + for terminated in [false, true] { + let mut oversized = std::io::repeat(b' ').take(LIMIT + 8192); + let tail: &[u8] = if terminated { b"\n{\"method\":\"notifications/initialized\"}\n" } else { b"" }; + let mut input = BufReader::with_capacity(256, (&mut oversized).chain(tail)); + let mut output = Vec::new(); + let mut server = Server::headless(); + server.serve(&mut input, &mut output).unwrap(); + assert!(!server.is_initialized(), "no input following the oversized frame is dispatched"); + assert!(oversized.limit() >= 8192 - 256, "stop after the limit plus one read buffer"); + let reply: Value = serde_json::from_slice(&output).unwrap(); + assert_eq!(reply["error"]["code"], -32600); + assert_eq!(reply["error"]["message"], "message too large"); + } +} + +#[test] +fn serve_accepts_a_limit_sized_frame_and_a_final_unterminated_frame() { + use std::io::{BufReader, Read}; + + const LIMIT: u64 = 32 * 1024 * 1024; + let ping = b"{\"id\":1,\"method\":\"ping\"}"; + let padding = std::io::repeat(b' ').take(LIMIT - ping.len() as u64); + let input = (&ping[..]).chain(padding).chain(&b"\n\r\n{\"id\":2,\"method\":\"ping\"}"[..]); + let mut output = Vec::new(); + Server::headless().serve(BufReader::new(input), &mut output).unwrap(); + let lines: Vec = String::from_utf8(output).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[0]["result"], json!({})); + assert_eq!(lines[1]["id"], 2); +} diff --git a/crates/ui-egui/src/control.rs b/crates/ui-egui/src/control.rs index 5b935a6..98fc9d8 100644 --- a/crates/ui-egui/src/control.rs +++ b/crates/ui-egui/src/control.rs @@ -42,6 +42,14 @@ pub struct ControlRequest { pub method: String, pub params: Value, pub reply: std::sync::mpsc::Sender, + /// The transport stops waiting after its timeout; discard requests still queued then. + pub cancelled: std::sync::Arc, +} + +impl ControlRequest { + pub fn is_cancelled(&self) -> bool { + self.cancelled.load(std::sync::atomic::Ordering::Relaxed) + } } /// Pending screenshot replies. @@ -87,6 +95,9 @@ pub fn drain(app: &mut FigApp, ctx: &Context) { } let reqs = std::mem::take(&mut app.control_queue); for r in reqs { + if r.is_cancelled() { + continue; + } // Requests for another window wait for its frame. let target = r.params.get("window").and_then(Value::as_u64).unwrap_or(app.windows.focused); if target != app.windows.current { diff --git a/crates/ui-egui/src/idle_tests.rs b/crates/ui-egui/src/idle_tests.rs index 917ab3f..2d4e575 100644 --- a/crates/ui-egui/src/idle_tests.rs +++ b/crates/ui-egui/src/idle_tests.rs @@ -227,14 +227,20 @@ fn control_requests_and_screenshots_wake_an_idle_ui() { h.settle(); // A request is answered in the next frame, which asks for one more to show its effect. let (tx, rx) = std::sync::mpsc::channel(); - h.app.control_queue.push(ControlRequest { id: json!(1), method: "ui.inspect".into(), params: json!({}), reply: tx.clone() }); + h.app.control_queue.push(ControlRequest { + id: json!(1), + method: "ui.inspect".into(), + params: json!({}), + reply: tx.clone(), + cancelled: Default::default(), + }); assert_eq!(h.frame(vec![]), Duration::ZERO); assert!(rx.try_recv().unwrap().result.is_ok()); h.assert_idle("after a control request"); // A screenshot keeps frames coming until the window delivers it, then the UI rests again. let path = std::env::temp_dir().join(format!("figcraft-idle-shot-{}.png", std::process::id())); let params = json!({"path": path.to_string_lossy()}); - h.app.control_queue.push(ControlRequest { id: json!(2), method: "ui.screenshot".into(), params, reply: tx }); + h.app.control_queue.push(ControlRequest { id: json!(2), method: "ui.screenshot".into(), params, reply: tx, cancelled: Default::default() }); for _ in 0..5 { assert_eq!(h.frame(vec![]), Duration::ZERO, "waiting for the screenshot"); } diff --git a/docs/control-protocol.md b/docs/control-protocol.md index 15c096d..bed647a 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -21,6 +21,20 @@ real UI state. A request that the UI doesn't answer within 60 s gets a timeout e not JSON gets `{"id": null, "ok": false, "error": "bad JSON: …"}`. The implementation is in `crates/ui-egui/src/control.rs` (methods) and `apps/figcraft/src/control_server.rs` (socket). +A request line is limited to **32 MiB before the newline**, including whitespace. The reader stops +at the first excess byte, replies `{"id": null, "ok": false, "error": "message too large"}`, then +closes the connection without consuming the rest of the frame. The server accepts at most **8 +active connections**; additional connections are closed immediately. Socket reads and writes each +have a **60 s inactivity timeout**, so idle clients and clients that stop reading release their +worker. This bounds blocked socket operations, not the total duration of a frame arriving slowly. +Clients can reconnect after an idle timeout (the attached MCP backend already does). + +Both the socket queue and the UI queue hold at most **8 requests** each, including requests waiting +for another window's frame. A full socket queue replies with `"control request queue is full"` +instead of dispatching the new request. These limits apply to the opt-in loopback control channel; +requests still queued after the UI timeout are discarded, including requests deferred to a hidden +window. A command already running is not interrupted. + Coordinates are page units, y down. `x`/`y` in `node.set`/`node.create` are relative to the parent frame (groups have no coordinate space of their own). diff --git a/docs/mcp.md b/docs/mcp.md index 77f286d..3dc12d0 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -5,6 +5,11 @@ design with it: create and edit layers, run any menu command, inspect the result it as a PNG. It speaks newline-delimited JSON-RPC 2.0 over stdio (MCP protocol `2025-06-18`; `2025-03-26` and `2024-11-05` clients are accepted too). +Incoming lines are limited to **32 MiB before the newline**, including whitespace. The limit is +enforced while reading: an oversized or unterminated frame stops at the first excess byte, returns +JSON-RPC error `-32600` (`message too large`, null id), and closes the stdio session without reading +the rest or dispatching later requests. Lines within the limit keep the same JSON-RPC behavior. + The server lives in `crates/mcp` (`figcraft-mcp`) and is started by the CLI: | Command | Backend |