Control channel: a timed-out request is cancelled, and the reply says whether it ran (fixes #345) (#348)
The control server gave up after 60 s with "timeout" but left the request queued, so the app ran it later and a client retry applied it twice. Requests now carry a shared state: on timeout the server cancels a request the app has not started (reply state "not-run", safe to retry) and the app skips it; one already started replies state "may-have-run". The MCP bridge passes that on in its error. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
45755697b6
commit
cdd25840cc
|
|
@ -42,12 +42,14 @@ fn serve(stream: TcpStream, tx: Sender<ControlRequest>, ctx: egui::Context) {
|
|||
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!({}));
|
||||
let (req, rrx) = ControlRequest::new(method, params);
|
||||
let (req, pending) = ControlRequest::new(method, params);
|
||||
if tx.send(req).is_err() {
|
||||
break;
|
||||
}
|
||||
ctx.request_repaint();
|
||||
let mut r = rrx.recv_timeout(Duration::from_secs(60)).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"}));
|
||||
// On timeout the request is cancelled if the app hasn't started it, and the reply's
|
||||
// `state` says whether it can still have run.
|
||||
let mut r = pending.wait(Duration::from_secs(60));
|
||||
if let Some(o) = r.as_object_mut() {
|
||||
o.insert("id".into(), id);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -98,7 +98,13 @@ impl Backend for Remote {
|
|||
if v.get("ok").and_then(Value::as_bool) == Some(true) {
|
||||
Ok(v.get("result").cloned().unwrap_or(Value::Null))
|
||||
} else {
|
||||
Err(v.get("error").and_then(Value::as_str).unwrap_or("error").to_string())
|
||||
let e = v.get("error").and_then(Value::as_str).unwrap_or("error");
|
||||
// A timeout says whether the request can still have run (#345).
|
||||
Err(match v.get("state").and_then(Value::as_str) {
|
||||
Some("not-run") => format!("{e}: CADCraft did not start `{method}` and won't; it is safe to send again"),
|
||||
Some("may-have-run") => format!("{e}: `{method}` may or may not have been applied. Inspect the drawing first"),
|
||||
_ => e.to_string(),
|
||||
})
|
||||
}
|
||||
}
|
||||
fn has_ui(&self) -> bool {
|
||||
|
|
@ -236,4 +242,24 @@ mod tests {
|
|||
assert!(err.contains("may or may not have been applied"), "{err}");
|
||||
assert_eq!(server.join().unwrap(), 1);
|
||||
}
|
||||
|
||||
/// The app's timeout reply says whether the request can still run; the error passes that on (#345).
|
||||
#[test]
|
||||
fn timeout_reply_says_whether_a_retry_is_safe() {
|
||||
for (state, want) in [("not-run", "safe to send again"), ("may-have-run", "Inspect the drawing first")] {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let addr = listener.local_addr().unwrap().to_string();
|
||||
let server = std::thread::spawn(move || {
|
||||
let (conn, _) = listener.accept().unwrap();
|
||||
let mut line = String::new();
|
||||
BufReader::new(conn.try_clone().unwrap()).read_line(&mut line).unwrap();
|
||||
let mut out = conn;
|
||||
writeln!(out, "{}", json!({"id": 1, "ok": false, "error": "timeout", "state": state})).unwrap();
|
||||
});
|
||||
let mut remote = Remote::connect(&addr).unwrap();
|
||||
let err = remote.call("engine.execute", json!({"command": "circle", "params": {}})).unwrap_err();
|
||||
assert!(err.starts_with("timeout") && err.contains(want), "{err}");
|
||||
server.join().unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,7 +18,10 @@
|
|||
//! render of the drawing (no window needed; fits the drawing unless `fit` is false; replies `pngBase64`)
|
||||
//! - `app.open {path}`, `app.save {path?}`, `app.quit`
|
||||
|
||||
use std::sync::mpsc::Sender;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU8, Ordering};
|
||||
use std::sync::mpsc::{Receiver, Sender};
|
||||
use std::time::Duration;
|
||||
|
||||
use cadcraft_engine::Input;
|
||||
use cadcraft_geom::Vec2;
|
||||
|
|
@ -32,12 +35,52 @@ pub struct ControlRequest {
|
|||
pub method: String,
|
||||
pub params: Value,
|
||||
pub reply: Sender<ControlResponse>,
|
||||
state: Arc<AtomicU8>,
|
||||
}
|
||||
|
||||
/// [`ControlRequest`] states, shared by the app and the transport waiting for the reply.
|
||||
const QUEUED: u8 = 0;
|
||||
const STARTED: u8 = 1;
|
||||
const CANCELLED: u8 = 2;
|
||||
|
||||
impl ControlRequest {
|
||||
pub fn new(method: impl Into<String>, params: Value) -> (Self, std::sync::mpsc::Receiver<ControlResponse>) {
|
||||
pub fn new(method: impl Into<String>, params: Value) -> (Self, PendingReply) {
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
(Self { method: method.into(), params, reply: tx }, rx)
|
||||
let state = Arc::new(AtomicU8::new(QUEUED));
|
||||
(Self { method: method.into(), params, reply: tx, state: state.clone() }, PendingReply { rx, state })
|
||||
}
|
||||
|
||||
/// Mark the request as started. False when the transport already gave up on it
|
||||
/// ([`PendingReply::wait`] timed out): the app must then skip it, since the client was told it
|
||||
/// did not run and may send it again.
|
||||
pub fn begin(&self) -> bool {
|
||||
self.state.compare_exchange(QUEUED, STARTED, Ordering::AcqRel, Ordering::Acquire).is_ok()
|
||||
}
|
||||
}
|
||||
|
||||
/// The transport's side of a [`ControlRequest`]: waits for the reply.
|
||||
pub struct PendingReply {
|
||||
rx: Receiver<ControlResponse>,
|
||||
state: Arc<AtomicU8>,
|
||||
}
|
||||
|
||||
impl PendingReply {
|
||||
/// The reply, or after `timeout` an error that says whether the request can still have
|
||||
/// changed anything:
|
||||
/// - `{"ok": false, "error": "timeout", "state": "not-run"}`: the app had not started it, and
|
||||
/// now never will. Sending it again is safe.
|
||||
/// - `{"ok": false, "error": "timeout", "state": "may-have-run"}`: the app had started it, so
|
||||
/// it may still finish and change the drawing (its late reply is discarded). Inspect before
|
||||
/// sending it again.
|
||||
pub fn wait(self, timeout: Duration) -> ControlResponse {
|
||||
if let Ok(r) = self.rx.recv_timeout(timeout) {
|
||||
return r;
|
||||
}
|
||||
if self.state.compare_exchange(QUEUED, CANCELLED, Ordering::AcqRel, Ordering::Acquire).is_ok() {
|
||||
return json!({"ok": false, "error": "timeout", "state": "not-run"});
|
||||
}
|
||||
// Started: a reply that arrived just now still counts.
|
||||
self.rx.try_recv().unwrap_or_else(|_| json!({"ok": false, "error": "timeout", "state": "may-have-run"}))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -356,6 +399,44 @@ mod tests {
|
|||
assert!((kept.scale - 5.0).abs() < 1e-9);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timed_out_request_that_never_started_is_skipped() {
|
||||
let (req, pending) = ControlRequest::new("engine.execute", json!({}));
|
||||
let r = pending.wait(Duration::from_millis(1));
|
||||
assert_eq!(r, json!({"ok": false, "error": "timeout", "state": "not-run"}));
|
||||
// The app reaches the request after the timeout: it must not run it.
|
||||
assert!(!req.begin());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timed_out_request_already_running_may_have_run() {
|
||||
let (req, pending) = ControlRequest::new("engine.execute", json!({}));
|
||||
assert!(req.begin());
|
||||
let r = pending.wait(Duration::from_millis(1));
|
||||
assert_eq!(r, json!({"ok": false, "error": "timeout", "state": "may-have-run"}));
|
||||
// The late result goes nowhere (the receiver is gone); sending it doesn't fail the app.
|
||||
assert!(req.reply.send(json!({"ok": true})).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reply_in_time_is_returned() {
|
||||
let (req, pending) = ControlRequest::new("engine.execute", json!({}));
|
||||
let app = std::thread::spawn(move || {
|
||||
assert!(req.begin());
|
||||
let _ = req.reply.send(json!({"ok": true, "result": 1}));
|
||||
});
|
||||
assert_eq!(pending.wait(Duration::from_secs(30)), json!({"ok": true, "result": 1}));
|
||||
let _ = app.join();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reply_sent_right_at_the_timeout_still_counts() {
|
||||
let (req, pending) = ControlRequest::new("engine.execute", json!({}));
|
||||
assert!(req.begin());
|
||||
let _ = req.reply.send(json!({"ok": true}));
|
||||
assert_eq!(pending.wait(Duration::ZERO), json!({"ok": true}));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn render_reply_has_base64_width_height_and_keeps_old_keys() {
|
||||
let r = render_reply(b"abc", 7, 5, Some("x.png"));
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ use cadcraft_engine::Session;
|
|||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{Value, json};
|
||||
|
||||
pub use control::{ControlRequest, ControlResponse};
|
||||
pub use control::{ControlRequest, ControlResponse, PendingReply};
|
||||
|
||||
/// The host storage key for [`CadApp::prefs_json`].
|
||||
pub const PREFS_KEY: &str = "cadcraft.prefs";
|
||||
|
|
@ -436,6 +436,10 @@ impl CadApp {
|
|||
fn drain_control(&mut self, ctx: &egui::Context) {
|
||||
let Some(rx) = self.control_rx.take() else { return };
|
||||
while let Ok(req) = rx.try_recv() {
|
||||
// Timed out while queued: the client was told it did not run (#345).
|
||||
if !req.begin() {
|
||||
continue;
|
||||
}
|
||||
let reply = req.reply.clone();
|
||||
match control::handle(self, ctx, &req) {
|
||||
control::Outcome::Done(v) => {
|
||||
|
|
|
|||
|
|
@ -10,6 +10,15 @@ Start the desktop app with `--control PORT` (or `CADCRAFT_CONTROL_PORT=PORT`). I
|
|||
|
||||
Errors come back as `{"id": …, "ok": false, "error": "message"}`.
|
||||
|
||||
The server waits up to 60 s for the app to answer a request. After that it replies with a timeout, and
|
||||
`state` tells whether sending the request again is safe:
|
||||
|
||||
- `{"id": …, "ok": false, "error": "timeout", "state": "not-run"}`: the app had not started the request.
|
||||
It is cancelled and will never run, so sending it again is safe.
|
||||
- `{"id": …, "ok": false, "error": "timeout", "state": "may-have-run"}`: the app had already started it.
|
||||
It may still finish and change the drawing, and its late result is discarded. Inspect the drawing
|
||||
(`drawing.inspect`) before sending it again.
|
||||
|
||||
| Method | Params | What it does |
|
||||
|---|---|---|
|
||||
| `cmdline.input` | `{text}` | Type a line at the command line exactly like a user. Starts commands by name/alias and answers prompts (points `x,y`, `@dx,dy`, `@d<a`, distances, keywords, empty = Enter). Spaces act as Enter except at text prompts. |
|
||||
|
|
|
|||
Loading…
Reference in New Issue