diff --git a/ROADMAP.md b/ROADMAP.md index 40407d18b6..b5aa6f28e3 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -135,6 +135,7 @@ Ranked, to beta (detail and remaining milestones in [docs/roadmap.md](docs/roadm ## Progress log | Date | What landed | +| 2026-10-16 | Control/MCP delivery (#1326, #1331): the MCP bridge never re-sends a mutation whose outcome is unknown (explicit "outcome unknown" error; reads and not-sent requests still retry); a control request whose caller timed out while queued never runs, and a timeout before execution is told apart from an unknown outcome; bounded MCP and reply line reads; capped unauthenticated control connections with fallible thread creation | | 2026-10-14 | DOCX row table border exceptions render and survive saving. Explicit cell borders win; a row exception does not change other rows. | | 2026-10-14 | Tab stops edited on the ruler: the corner selector picks Left/Center/Right/Decimal/Bar, a click sets a stop (Alt skips the snap), dragging moves it with a guide line and position tooltip, dragging off the ruler clears it; markers draw per kind and mirror for RTL; one `para.tabs` call and one undo step per gesture | | 2026-10-12 | Fields keep `w:dirty` through DOCX saves (#877): a begin marker flagged by a generator that couldn't compute a result (pandoc-style TOC placeholders) keeps the flag on open/save — as a buffered field or a range field — so Word still refreshes it on open; computing a result (Update Field, Update Table) clears it, like Word's F9 | diff --git a/apps/wordcraft/src/control_server.rs b/apps/wordcraft/src/control_server.rs index 2b8f4137d7..353331f2bd 100644 --- a/apps/wordcraft/src/control_server.rs +++ b/apps/wordcraft/src/control_server.rs @@ -9,6 +9,7 @@ use std::io::{BufRead, BufReader, Read, Write}; use std::net::{TcpListener, TcpStream}; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc::{Receiver, Sender, channel}; use std::time::{Duration, Instant}; @@ -21,6 +22,38 @@ const MAX_LINE: usize = 1 << 20; /// A connection that sends no authorised request in this time is closed. const FIRST_REQUEST: Duration = Duration::from_secs(30); +/// How long a caller waits for the UI to answer a request. A request still queued after this +/// is cancelled and never runs. +const REPLY_WAIT: Duration = Duration::from_secs(60); + +/// Connections that have not yet sent an authorised request are limited to this many; more are +/// closed at once. Each one holds a thread until its first request or [`FIRST_REQUEST`]. +const MAX_UNAUTHENTICATED: usize = 64; + +/// Counts one unauthenticated connection until it authenticates or ends. +struct Admission(Option>); + +impl Admission { + fn enter(count: &Arc, max: usize) -> Option { + if count.fetch_add(1, Ordering::SeqCst) >= max { + count.fetch_sub(1, Ordering::SeqCst); + return None; + } + Some(Self(Some(count.clone()))) + } + fn release(&mut self) { + if let Some(c) = self.0.take() { + c.fetch_sub(1, Ordering::SeqCst); + } + } +} + +impl Drop for Admission { + fn drop(&mut self) { + self.release(); + } +} + /// The key file of a running control server; [`KeyFile::remove`] it when the window closes. pub struct KeyFile(PathBuf); @@ -104,14 +137,37 @@ fn keys_match(given: &str, key: &str) -> bool { } fn serve_on(listener: TcpListener, ctx: egui::Context, key: String, first: Duration) -> Receiver { + serve_with(listener, ctx, key, Limits { first, wait: REPLY_WAIT, max_unauthenticated: MAX_UNAUTHENTICATED }) +} + +/// The time and count budgets of a server (small ones in tests). +#[derive(Clone, Copy)] +struct Limits { + first: Duration, + wait: Duration, + max_unauthenticated: usize, +} + +fn serve_with(listener: TcpListener, ctx: egui::Context, key: String, limits: Limits) -> Receiver { let (tx, rx) = channel::(); let key: Arc = key.into(); - std::thread::spawn(move || { + let unauthenticated = Arc::new(AtomicUsize::new(0)); + let accept = move || { for stream in listener.incoming().flatten() { + let Some(admission) = Admission::enter(&unauthenticated, limits.max_unauthenticated) else { + log::warn!("control server: too many unauthenticated connections, one closed"); + continue; + }; let (tx, ctx, key) = (tx.clone(), ctx.clone(), key.clone()); - std::thread::spawn(move || serve(stream, tx, ctx, &key, first)); + // A thread that cannot start closes this connection only (the guard releases its slot). + if let Err(e) = std::thread::Builder::new().name("control-conn".into()).spawn(move || serve(stream, tx, ctx, &key, limits, admission)) { + log::warn!("control server: no thread for a connection: {e}"); + } } - }); + }; + if let Err(e) = std::thread::Builder::new().name("control-accept".into()).spawn(accept) { + log::error!("control server: cannot start the accept thread: {e}"); + } rx } @@ -175,11 +231,11 @@ fn discard_rest_of_line(r: &mut BufReader) { } } -fn serve(stream: TcpStream, tx: Sender, ctx: egui::Context, key: &str, first: Duration) { +fn serve(stream: TcpStream, tx: Sender, ctx: egui::Context, key: &str, limits: Limits, mut admission: Admission) { let Ok(read) = stream.try_clone() else { return }; let mut out = stream; // One overall deadline from accept to the first authorised request; after it, none. - let mut reader = BufReader::new(Timed { s: read, deadline: Some(Instant::now() + first) }); + let mut reader = BufReader::new(Timed { s: read, deadline: Some(Instant::now() + limits.first) }); let mut authorised = false; // The key an `auth` request gave this connection. let mut conn_key: Option = None; @@ -228,6 +284,7 @@ fn serve(stream: TcpStream, tx: Sender, ctx: egui::Context, key: } if !authorised { authorised = true; + admission.release(); reader.get_mut().set_deadline(None); } if method == "auth" { @@ -237,12 +294,18 @@ fn serve(stream: TcpStream, tx: Sender, ctx: egui::Context, key: } continue; } - let (req, rrx) = ControlRequest::new(method, params); + let (req, rrx, waiter) = ControlRequest::with_waiter(method, params); if tx.send(req).is_err() { break; } ctx.request_repaint(); - let r = rrx.recv_timeout(Duration::from_secs(60)).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"})); + let r = rrx.recv_timeout(limits.wait).unwrap_or_else(|_| { + if waiter.cancel() { + json!({"ok": false, "error": "timeout before execution: the request was not run and never will be"}) + } else { + json!({"ok": false, "error": "timeout: outcome unknown, the request had started when the wait ended; inspect the document to check"}) + } + }); if !reply(r) { break; } @@ -532,4 +595,52 @@ mod tests { } let _ = std::fs::remove_dir_all(&dir); } + + fn boot_limits(limits: Limits) -> (u16, Receiver) { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let port = listener.local_addr().unwrap().port(); + (port, serve_with(listener, egui::Context::default(), KEY.to_string(), limits)) + } + + fn small(wait_ms: u64, max_unauthenticated: usize) -> Limits { + Limits { first: FIRST_REQUEST, wait: Duration::from_millis(wait_ms), max_unauthenticated } + } + + #[test] + fn a_request_that_timed_out_in_the_queue_never_runs() { + let (port, rx) = boot_limits(small(150, 8)); + let mut c = Conn::open(port); + let r = c.call(&format!(r#"{{"id":1,"key":"{KEY}","method":"engine.execute","params":{{"command":"text.insert"}}}}"#)); + assert_eq!(r["ok"], false); + assert!(r["error"].as_str().unwrap_or("").starts_with("timeout before execution"), "{r}"); + // The UI gets to it late: it must not run it. + let req = rx.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!(!req.start(), "a cancelled request was started"); + } + + #[test] + fn a_request_that_started_before_the_timeout_reports_an_unknown_outcome() { + let (port, rx) = boot_limits(small(150, 8)); + let mut c = Conn::open(port); + c.send(&format!(r#"{{"id":1,"key":"{KEY}","method":"engine.execute","params":{{"command":"text.insert"}}}}"#)); + let req = rx.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!(req.start()); + let r = c.recv(); + assert!(r["error"].as_str().unwrap_or("").contains("outcome unknown"), "{r}"); + } + + #[test] + fn unauthenticated_connections_are_limited() { + let (port, rx) = boot_limits(small(5_000, 2)); + let mut idle = [Conn::open(port), Conn::open(port)]; + // Give the server time to count both. + std::thread::sleep(Duration::from_millis(200)); + assert!(Conn::open(port).closed(), "the third idle connection was admitted"); + // One authenticates: its slot is free again. + assert_eq!(idle[0].call(&format!(r#"{{"id":1,"method":"auth","key":"{KEY}"}}"#))["ok"], true); + std::thread::sleep(Duration::from_millis(200)); + let mut again = Conn::open(port); + assert_eq!(again.call(&format!(r#"{{"id":1,"method":"auth","key":"{KEY}"}}"#))["ok"], true); + assert!(rx.try_recv().is_err()); + } } diff --git a/crates/mcp/src/backend.rs b/crates/mcp/src/backend.rs index ac32f5d32b..8101a46ba8 100644 --- a/crates/mcp/src/backend.rs +++ b/crates/mcp/src/backend.rs @@ -1,6 +1,6 @@ //! Where MCP tool calls end up: a control-channel method call. -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; use std::net::{IpAddr, SocketAddr, TcpStream, ToSocketAddrs}; use std::path::Path; use std::time::Duration; @@ -141,27 +141,29 @@ impl Remote { Ok(req.to_string()) } - fn roundtrip(&mut self, line: &str) -> std::io::Result { + /// Send one line and read the reply line. A failure before the whole line (with its newline) + /// reached the socket is [`Failure::NotSent`]: the app cannot have run it. Any failure after + /// that is [`Failure::Unknown`]: the app may have run the request and lost the reply. + fn roundtrip(&mut self, line: &str) -> Result { let Some((reader, writer)) = self.conn.as_mut() else { - return Err(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected")); + return Err(Failure::NotSent(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected"))); }; - writer.write_all(line.as_bytes())?; - writer.write_all(b"\n")?; - writer.flush()?; - let mut reply = String::new(); - if reader.read_line(&mut reply)? == 0 { - return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "control channel closed")); - } - Ok(reply) + let mut out = Vec::with_capacity(line.len() + 1); + out.extend_from_slice(line.as_bytes()); + out.push(b'\n'); + writer.write_all(&out).and_then(|()| writer.flush()).map_err(Failure::NotSent)?; + let reply = read_bounded_line(reader, MAX_REPLY).map_err(Failure::Unknown)?; + String::from_utf8(reply).map_err(|_| Failure::Unknown(std::io::Error::new(std::io::ErrorKind::InvalidData, "reply is not UTF-8"))) } - /// One request and its reply, with one retry on a new connection (the app may have - /// restarted; a new connection also reads the key again). + /// One request and its reply. A request that was not sent, or one that only reads, is tried + /// once more on a new connection (the app may have restarted; a new connection also reads + /// the key again). A mutation whose outcome is unknown is never sent again: it may have run. fn exchange(&mut self, id: u64, method: &str, params: &Value) -> Result { let unreachable = |addr: &str, e: std::io::Error| format!("WordCraft app at {addr} is not reachable: {e}"); let mut first = true; let reply = loop { - let tried = if self.conn.is_some() { Ok(()) } else { self.reconnect() }; + let tried = if self.conn.is_some() { Ok(()) } else { self.reconnect().map_err(Failure::NotSent) }; let result = match tried { Ok(()) => { let line = self.request_line(id, method, params)?; @@ -171,10 +173,18 @@ impl Remote { }; match result { Ok(r) => break r, - Err(e) => { + Err(f) => { self.conn = None; + if let Failure::Unknown(e) = &f + && !is_read_only(method) + { + return Err(format!( + "outcome unknown: the connection to the WordCraft app at {} failed after `{method}` was sent ({e}); it may have been executed, so it is not sent again. Inspect the document to check", + self.addr + )); + } if !first { - return Err(unreachable(&self.addr, e)); + return Err(unreachable(&self.addr, f.into_error())); } first = false; } @@ -184,6 +194,44 @@ impl Remote { } } +/// Why a round trip failed. +enum Failure { + /// Before the whole request reached the socket: the app did not run it. + NotSent(std::io::Error), + /// After the request was sent: the app may have run it. + Unknown(std::io::Error), +} + +impl Failure { + fn into_error(self) -> std::io::Error { + match self { + Self::NotSent(e) | Self::Unknown(e) => e, + } + } +} + +/// Longest reply line accepted from the app (screenshots and renders come as base64). +const MAX_REPLY: usize = 128 << 20; + +/// Methods that only read: sending them twice does no harm. +fn is_read_only(method: &str) -> bool { + matches!(method, "document.inspect" | "engine.commands" | "ui.inspect" | "ui.parity" | "ui.render" | "ui.screenshot") +} + +/// One line (with its newline) of at most `max` bytes. An end of stream before any byte, or a +/// longer line, is an error. +fn read_bounded_line(r: &mut impl BufRead, max: usize) -> std::io::Result> { + let mut buf = Vec::new(); + let n = Read::take(&mut *r, max as u64 + 1).read_until(b'\n', &mut buf)?; + if n == 0 { + return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "control channel closed")); + } + if buf.len() > max && !buf.ends_with(b"\n") { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, format!("reply line longer than {max} bytes"))); + } + Ok(buf) +} + impl Backend for Remote { fn call(&mut self, method: &str, params: Value) -> Result { let id = self.next_id; @@ -382,4 +430,44 @@ mod tests { assert!(e.contains("not a loopback address") && e.contains("not sent"), "{e}"); assert!(seen.recv_timeout(WAIT).unwrap().get("key").is_none()); } + + #[test] + fn a_mutation_with_a_lost_reply_is_not_sent_again() { + // The app runs the request, then the connection drops without a reply. + let (port, seen) = fake_app(|_| None); + let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap(); + let e = r.call("engine.execute", json!({"command": "text.insert", "params": {"text": "x"}})).unwrap_err(); + assert!(e.contains("outcome unknown"), "{e}"); + assert_eq!(seen.recv_timeout(WAIT).unwrap()["method"], "engine.execute"); + std::thread::sleep(Duration::from_millis(200)); + assert!(seen.try_recv().is_err(), "the mutation was sent a second time"); + } + + #[test] + fn a_read_with_a_lost_reply_is_sent_again() { + let n = Arc::new(AtomicUsize::new(0)); + let (port, seen) = fake_app(move |_| if n.fetch_add(1, Ordering::SeqCst) == 0 { None } else { Some((json!({"ok": true, "result": 3}), false)) }); + let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap(); + assert_eq!(r.call("document.inspect", json!({})).unwrap(), json!(3)); + assert_eq!(seen.try_iter().count(), 2); + } + + #[test] + fn a_mutation_not_sent_goes_out_on_a_new_connection() { + let (port, seen) = fake_app(|_| Some((json!({"ok": true, "result": 1}), false))); + let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap(); + r.conn = None; + assert_eq!(r.call("engine.execute", json!({"command": "file.new"})).unwrap(), json!(1)); + assert_eq!(seen.try_iter().count(), 1); + } + + #[test] + fn a_reply_line_is_bounded() { + let mut ok = std::io::Cursor::new(b"abc\nrest".to_vec()); + assert_eq!(read_bounded_line(&mut ok, 8).unwrap(), b"abc\n"); + let mut long = std::io::Cursor::new(vec![b'x'; 100]); + assert!(read_bounded_line(&mut long, 8).is_err()); + let mut empty = std::io::Cursor::new(Vec::new()); + assert!(read_bounded_line(&mut empty, 8).is_err()); + } } diff --git a/crates/mcp/src/server.rs b/crates/mcp/src/server.rs index 63f6764c61..2578ba1553 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 methods. -use std::io::{BufRead, Write}; +use std::io::{BufRead, Read, Write}; use serde_json::{Value, json}; @@ -24,6 +24,42 @@ parameters with list_commands and run them with execute (or several with batch). \"Hello\" → execute text.newParagraph → execute para.style {style:\"Heading 1\"} → select_text {text:\"Hello\"} → \ execute format.bold → inspect_document to verify → render_page to look at the result → save {path:\"out.docx\"}."; +/// Longest request line accepted on stdin. +const MAX_LINE: usize = 32 << 20; + +enum Line { + End, + TooLong, + Text(String), +} + +/// The next line of at most `max` bytes. A longer one is read to its end and dropped, a line +/// that is not UTF-8 is replaced lossily (the JSON parser then rejects it). +fn read_line(input: &mut impl BufRead, max: usize) -> std::io::Result { + let mut buf = Vec::new(); + let n = Read::take(&mut *input, max as u64 + 1).read_until(b'\n', &mut buf)?; + if n == 0 { + return Ok(Line::End); + } + if buf.len() > max && !buf.ends_with(b"\n") { + loop { + let (used, done) = match input.fill_buf()? { + [] => break, + b => match b.iter().position(|&c| c == b'\n') { + Some(i) => (i + 1, true), + None => (b.len(), false), + }, + }; + input.consume(used); + if done { + break; + } + } + return Ok(Line::TooLong); + } + Ok(Line::Text(String::from_utf8_lossy(&buf).into_owned())) +} + /// Resource URIs. pub const DOC_URI: &str = "wordcraft://document"; pub const COMMANDS_URI: &str = "wordcraft://commands"; @@ -58,16 +94,25 @@ impl Server { /// Serve newline-delimited JSON-RPC until `input` closes. Logs go to stderr only (stdout is /// the protocol stream). - pub fn serve(&mut self, input: impl BufRead, mut output: impl Write) -> std::io::Result<()> { - for line in input.lines() { - let line = line?; - if let Some(reply) = self.handle_line(&line) { + pub fn serve(&mut self, input: impl BufRead, output: impl Write) -> std::io::Result<()> { + self.serve_bounded(input, output, MAX_LINE) + } + + /// [`Server::serve`] with a request line limit of `max` bytes. A longer line is dropped + /// (without keeping it in memory) and answered with an error. + fn serve_bounded(&mut self, mut input: impl BufRead, mut output: impl Write, max: usize) -> std::io::Result<()> { + loop { + let reply = match read_line(&mut input, max)? { + Line::End => return Ok(()), + Line::TooLong => Some(error(Value::Null, INVALID_REQUEST, format!("request line longer than {max} bytes")).to_string()), + Line::Text(line) => self.handle_line(&line), + }; + if let Some(reply) = reply { output.write_all(reply.as_bytes())?; output.write_all(b"\n")?; output.flush()?; } } - Ok(()) } /// Handle one line; returns the reply line (None for notifications and blank lines). @@ -166,3 +211,40 @@ impl Server { } } } + +#[cfg(test)] +mod bounded_tests { + use super::*; + + struct Null; + impl Backend for Null { + fn call(&mut self, _: &str, _: Value) -> Result { + Ok(Value::Null) + } + fn has_ui(&self) -> bool { + false + } + fn describe(&self) -> String { + "test".into() + } + } + + #[test] + fn an_oversize_line_is_refused_and_the_next_one_still_served() { + let mut input = vec![b'x'; 200]; + input.extend_from_slice(b"\n{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"ping\"}\n"); + let mut out = Vec::new(); + Server::new(Box::new(Null)).serve_bounded(std::io::Cursor::new(input), &mut out, 64).unwrap(); + let lines: Vec = String::from_utf8(out).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect(); + assert_eq!(lines.len(), 2, "{lines:?}"); + assert_eq!(lines[0]["error"]["code"], INVALID_REQUEST); + assert_eq!(lines[1]["id"], 1); + } + + #[test] + fn a_line_without_a_newline_is_bounded() { + let mut out = Vec::new(); + Server::new(Box::new(Null)).serve_bounded(std::io::Cursor::new(vec![b'x'; 10_000]), &mut out, 64).unwrap(); + assert!(String::from_utf8(out).unwrap().contains("longer than 64")); + } +} diff --git a/crates/ui-egui/src/control.rs b/crates/ui-egui/src/control.rs index 9e09e54fee..9da69d469f 100644 --- a/crates/ui-egui/src/control.rs +++ b/crates/ui-egui/src/control.rs @@ -23,12 +23,45 @@ pub struct ControlRequest { pub method: String, pub params: Value, pub reply: Sender, + /// Shared with the waiter: whichever of "start" and "give up" happens first wins. + state: std::sync::Arc, +} + +const QUEUED: u8 = 0; +const STARTED: u8 = 1; +const CANCELLED: u8 = 2; + +/// The waiter's side of a [`ControlRequest`]: it gives up on a request that has not started. +#[derive(Clone)] +pub struct ControlWaiter(std::sync::Arc); + +impl ControlWaiter { + /// Cancel the request. True when it had not started: it will never run. False when the UI + /// already started it: its outcome is unknown to the waiter. + pub fn cancel(&self) -> bool { + use std::sync::atomic::Ordering::SeqCst; + self.0.compare_exchange(QUEUED, CANCELLED, SeqCst, SeqCst).is_ok() + } } impl ControlRequest { pub fn new(method: impl Into, params: Value) -> (Self, std::sync::mpsc::Receiver) { + let (req, rx, _) = Self::with_waiter(method, params); + (req, rx) + } + + /// Like [`ControlRequest::new`], plus the handle the waiter uses to cancel a queued request. + pub fn with_waiter(method: impl Into, params: Value) -> (Self, std::sync::mpsc::Receiver, ControlWaiter) { let (tx, rx) = std::sync::mpsc::channel(); - (Self { method: method.into(), params, reply: tx }, rx) + let state = std::sync::Arc::new(std::sync::atomic::AtomicU8::new(QUEUED)); + (Self { method: method.into(), params, reply: tx, state: state.clone() }, rx, ControlWaiter(state)) + } + + /// The UI is about to run the request. False when the waiter already gave up on it: it + /// must not run. + pub fn start(&self) -> bool { + use std::sync::atomic::Ordering::SeqCst; + self.state.compare_exchange(QUEUED, STARTED, SeqCst, SeqCst).is_ok() } } diff --git a/crates/ui-egui/src/lib.rs b/crates/ui-egui/src/lib.rs index 662516fbd6..5d8ed722b5 100644 --- a/crates/ui-egui/src/lib.rs +++ b/crates/ui-egui/src/lib.rs @@ -2159,6 +2159,11 @@ impl WordApp { let Some(rx) = self.control_rx.take() else { return }; while let Ok(req) = rx.try_recv() { let reply = req.reply.clone(); + if !req.start() { + // The caller timed out while this waited in the queue and was told it did not run. + log::debug!("control request {} dropped: its caller gave up", req.method); + continue; + } match control::handle(self, ctx, &req) { control::Outcome::Done(v) => { let _ = reply.send(v);