M-ctl: MCP bridge never replays a mutation of unknown outcome; queued control requests honour the caller's timeout (#1326, #1331)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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 |

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<Arc<AtomicUsize>>);
+
+impl Admission {
+    fn enter(count: &Arc<AtomicUsize>, max: usize) -> Option<Self> {
+        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<ControlRequest> {
+    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<ControlRequest> {
     let (tx, rx) = channel::<ControlRequest>();
     let key: Arc<str> = 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<Timed>) {
     }
 }
 
-fn serve(stream: TcpStream, tx: Sender<ControlRequest>, ctx: egui::Context, key: &str, first: Duration) {
+fn serve(stream: TcpStream, tx: Sender<ControlRequest>, 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<String> = None;
@@ -228,6 +284,7 @@ fn serve(stream: TcpStream, tx: Sender<ControlRequest>, 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<ControlRequest>, 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<ControlRequest>) {
+        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());
+    }
 }

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<String> {
+    /// 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<String, Failure> {
         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<Value, String> {
         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<Vec<u8>> {
+    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<Value, String> {
         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());
+    }
 }

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<Line> {
+    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<Value, String> {
+            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<Value> = 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"));
+    }
+}

crates/ui-egui/src/control.rs

@@ -23,12 +23,45 @@ pub struct ControlRequest {
     pub method: String,
     pub params: Value,
     pub reply: Sender<Value>,
+    /// Shared with the waiter: whichever of "start" and "give up" happens first wins.
+    state: std::sync::Arc<std::sync::atomic::AtomicU8>,
+}
+
+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<std::sync::atomic::AtomicU8>);
+
+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<String>, params: Value) -> (Self, std::sync::mpsc::Receiver<Value>) {
+        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<String>, params: Value) -> (Self, std::sync::mpsc::Receiver<Value>, 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()
     }
 }
 

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);