diff --git a/apps/solidcraft/src/control_server.rs b/apps/solidcraft/src/control_server.rs index 2e566afc..8680d0cd 100644 --- a/apps/solidcraft/src/control_server.rs +++ b/apps/solidcraft/src/control_server.rs @@ -4,7 +4,7 @@ use std::io::{BufRead, BufReader, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::mpsc::{Receiver, Sender, channel}; +use std::sync::mpsc::{Receiver, Sender, TryRecvError, channel}; use std::time::Duration; use serde_json::{Value, json}; @@ -160,12 +160,35 @@ fn refuse(out: &mut TcpStream, reader: &mut BufReader) { let _ = std::io::copy(&mut reader.by_ref().take(64 * 1024), &mut std::io::sink()); } +/// How long a request waits for the app's reply: most requests, and a batch (or every request, +/// on a patient connection). +#[derive(Clone, Copy)] +struct Waits { + short: Duration, + long: Duration, +} + +const WAITS: Waits = Waits { short: Duration::from_secs(60), long: Duration::from_secs(3600) }; + +/// A request that timed out but is still running in the app (#1618): its id and method, and where +/// its reply will land. +struct Pending { + id: Value, + method: String, + reply: Receiver, +} + fn serve(stream: TcpStream, token: &str, tx: Sender, ctx: egui::Context, patient: bool) { + serve_with(stream, token, tx, ctx, patient, WAITS); +} + +fn serve_with(stream: TcpStream, token: &str, tx: Sender, ctx: egui::Context, patient: bool, waits: Waits) { let Ok(read) = stream.try_clone() else { return }; let mut out = stream; let mut reader = BufReader::new(read); let mut bad = 0usize; let mut authenticated = false; + let mut pending: Option = None; loop { let line = match read_line(&mut reader, MAX_LINE) { Ok(Some(l)) => String::from_utf8_lossy(&l).into_owned(), @@ -193,14 +216,46 @@ fn serve(stream: TcpStream, token: &str, tx: Sender, ctx: egui:: let id = msg.get("id").cloned().unwrap_or(Value::Null); let method = msg.get("method").and_then(Value::as_str).unwrap_or("").to_string(); let params = msg.get("params").cloned().unwrap_or(json!({})); - // A batch replaces many short requests: the whole of it may take long (#743). - let wait = Duration::from_secs(if patient || method == "engine.batch" { 3600 } else { 60 }); - let (req, rrx) = ControlRequest::new(method, params); - if tx.send(req).is_err() { - break; - } - ctx.request_repaint(); - let mut r = rrx.recv_timeout(wait).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"})); + // A request that timed out may have finished since (#1618). + let late = pending.as_ref().and_then(|p| match p.reply.try_recv() { + Ok(reply) => Some(reply), + Err(TryRecvError::Disconnected) => Some(json!({"ok": false, "error": "the app ended the request without a reply"})), + Err(TryRecvError::Empty) => None, + }); + let mut r = if method == "control.pending" { + match (pending.take(), late) { + (None, _) => json!({"ok": true, "result": {"pending": null}}), + (Some(p), Some(reply)) => json!({"ok": true, "result": {"pending": p.id, "method": p.method, "done": true, "reply": reply}}), + (Some(p), None) => { + let r = json!({"ok": true, "result": {"pending": p.id, "method": p.method, "done": false}}); + pending = Some(p); + r + } + } + } else if let Some(p) = pending.as_ref().filter(|_| late.is_none()) { + // Nothing more is queued behind it, to run unseen when it ends. + json!({"ok": false, "busy": true, "error": format!("busy: request {} (`{}`) is still running; ask `control.pending` for its result", p.id, p.method)}) + } else { + if late.is_some() { + // Finished, and the client moved on without asking: it is dropped. + pending = None; + } + // A batch replaces many short requests: the whole of it may take long (#743). + let wait = if patient || method == "engine.batch" { waits.long } else { waits.short }; + let (req, rrx) = ControlRequest::new(method.clone(), params); + if tx.send(req).is_err() { + break; + } + ctx.request_repaint(); + match rrx.recv_timeout(wait) { + Ok(r) => r, + Err(_) => { + let r = json!({"ok": false, "running": true, "error": format!("timeout: `{method}` is still running in the app and will still apply; ask `control.pending` for its result")}); + pending = Some(Pending { id: id.clone(), method, reply: rrx }); + r + } + } + }; if let Some(o) = r.as_object_mut() { o.insert("id".into(), id); } @@ -228,6 +283,62 @@ fn serve(stream: TcpStream, token: &str, tx: Sender, ctx: egui:: mod tests { use super::*; + /// #1618: a request that outlasts its wait says it is still running; the connection keeps its + /// reply for `control.pending`, and refuses other requests as `busy` until it lands, so none + /// is queued to run unseen behind it. + #[test] + fn a_timed_out_request_can_be_asked_for_later() { + let token = "0123456789abcdef0123456789abcdef"; + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let (tx, rx) = channel::(); + let ran = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); + let seen = ran.clone(); + std::thread::spawn(move || { + for req in rx { + seen.lock().unwrap().push(req.method.clone()); + if req.method == "slow" { + std::thread::sleep(Duration::from_millis(400)); + } + let _ = req.reply.send(json!({"ok": true, "result": req.method})); + } + }); + let waits = Waits { short: Duration::from_millis(100), long: Duration::from_millis(100) }; + std::thread::spawn(move || { + if let Some(stream) = listener.incoming().flatten().next() { + serve_with(stream, token, tx, egui::Context::default(), false, waits); + } + }); + let s = TcpStream::connect(addr).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(10))).unwrap(); + let mut out = s.try_clone().unwrap(); + let mut lines = BufReader::new(s); + let mut ask = |v: Value| -> Value { + writeln!(out, "{v}").unwrap(); + let mut l = String::new(); + lines.read_line(&mut l).unwrap(); + serde_json::from_str(&l).unwrap() + }; + let r = ask(json!({"id": 1, "method": "slow", "token": token})); + assert!(r["ok"] == false && r["running"] == true && r["error"].as_str().is_some_and(|e| e.starts_with("timeout")), "{r}"); + let r = ask(json!({"id": 2, "method": "ui.view"})); + assert!(r["busy"] == true && r["id"] == 2, "{r}"); + let r = ask(json!({"id": 3, "method": "control.pending"})); + assert_eq!((&r["result"]["pending"], &r["result"]["done"]), (&json!(1), &json!(false)), "{r}"); + std::thread::sleep(Duration::from_millis(600)); + let r = ask(json!({"id": 4, "method": "control.pending"})); + assert_eq!( + (&r["result"]["pending"], &r["result"]["done"], &r["result"]["reply"]["result"]), + (&json!(1), &json!(true), &json!("slow")), + "{r}" + ); + let r = ask(json!({"id": 5, "method": "control.pending"})); + assert_eq!(r["result"]["pending"], Value::Null, "{r}"); + let r = ask(json!({"id": 6, "method": "ui.view"})); + assert_eq!((&r["ok"], &r["result"]), (&json!(true), &json!("ui.view")), "{r}"); + assert_eq!(*ran.lock().unwrap(), ["slow", "ui.view"], "the busy request never reached the app"); + } + /// #205: a connection runs nothing until a request carries the launch's token; a wrong or /// missing token, or garbage, is refused and the connection closed. Once shown, the token /// need not be repeated on that connection. diff --git a/docs/control-protocol.md b/docs/control-protocol.md index 142ea263..ebddf4c3 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -48,7 +48,16 @@ connections that send eight lines that are not JSON. At most 16 connections are (further ones get `{"ok": false, "error": "too many connections"}`), at most 64 queued requests run per frame so a flood cannot starve drawing, and a request that the UI thread has not answered within 60 seconds (3600 seconds for `engine.batch`) is replied to with -`{"ok": false, "error": "timeout"}`. The same 16 MiB line limit applies to the MCP server. +`{"ok": false, "running": true, "error": "timeout: … is still running …"}`. The same 16 MiB line +limit applies to the MCP server. + +A timed-out request has not failed: it is still running in the app and will still apply (a long +fillet, say), so don't simply send it again. The connection keeps its reply (#1618). Ask for it +with `control.pending`: `{"pending": , "method": …, "done": false}` while it runs, then +`{"pending": , "method": …, "done": true, "reply": {…}}` (its own reply, once), and +`{"pending": null}` when nothing is outstanding. Until it lands, any other request on that +connection is refused at once with `{"ok": false, "busy": true, "error": "busy: …"}` and not +run, so nothing is queued to apply unseen behind it. Any request's `params` may carry `allow_paths: ["/abs/folder", …]`: every file that request reads, writes or deletes (command paths such as `file.export`'s `path` or `doc.recovery_*`'s