control: a request that timed out while queued is abandoned, never run late (#719)

This commit is contained in:
midasmatterhorn
2026-10-15 00:37:58 +03:00
parent ea904591a5
commit 2568c55afa
4 changed files with 104 additions and 3 deletions
+53 -1
View File
@@ -151,10 +151,23 @@ fn serve(stream: TcpStream, tx: &SyncSender<ControlRequest>, ctx: &egui::Context
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 state = Arc::clone(&req.state);
let mut r = match tx.try_send(req) {
Ok(()) => {
ctx.request_repaint();
rrx.recv_timeout(limits.reply_timeout).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"}))
match rrx.recv_timeout(limits.reply_timeout) {
Ok(v) => v,
Err(_) => {
// Give up on the request. While it was still queued this makes
// sure it never runs late (#719); if the app had already taken
// it, its outcome is unknown rather than a definite failure.
if ControlRequest::abandon(&state) {
json!({"ok": false, "error": "timeout"})
} else {
json!({"ok": false, "error": "timeout: the app had already started the request, so it may still run; its outcome is unknown"})
}
}
}
}
Err(TrySendError::Full(_)) => json!({"ok": false, "error": "busy: too many queued requests"}),
Err(TrySendError::Disconnected(_)) => break,
@@ -296,4 +309,43 @@ mod tests {
assert_eq!(a["error"], "timeout");
drop(rx);
}
/// A request that timed out while it was still queued must not run when the window resumes (#719).
#[test]
fn timed_out_request_does_not_run_when_the_window_resumes() {
let (addr, rx) = server(Limits { max_line: 512, ..small() });
let (mut r, mut w) = client(addr);
writeln!(w, r#"{{"id":1,"method":"engine.execute","params":{{"command":"slide.new"}}}}"#).unwrap();
let t = reply(&mut r).unwrap();
assert_eq!(t["error"], "timeout");
// The window resumes and drains the queue, exactly as `SlideApp::drain_control` does.
let mut app = deckcraft_ui_egui::SlideApp::new(deckcraft_engine::Session::with_new(), deckcraft_ui_egui::Services::default());
let ctx = egui::Context::default();
let before = app.session.active().unwrap().doc.slides.len();
let req = rx.try_recv().expect("the timed-out request is still queued");
if req.take() {
let _ = deckcraft_ui_egui::control::handle(&mut app, &ctx, &req);
}
assert_eq!(app.session.active().unwrap().doc.slides.len(), before, "a request that timed out must not run when the window resumes");
drop(rx);
}
/// A request the app had already started when the reply timed out reports an unknown outcome,
/// not a definite failure (#719).
#[test]
fn a_request_already_started_when_the_reply_times_out_reports_an_unknown_outcome() {
let (addr, rx) = server(small());
let ui = std::thread::spawn(move || {
let req = rx.recv().expect("the request is queued");
assert!(req.take(), "the app claims the request");
// Still running it when the sender's wait expires.
std::thread::sleep(Duration::from_millis(450));
let _ = req.reply.send(json!({"ok": true, "result": "ran"}));
});
let (mut r, mut w) = client(addr);
writeln!(w, r#"{{"id":1,"method":"m"}}"#).unwrap();
let v = reply(&mut r).unwrap();
assert!(v["error"].as_str().unwrap().contains("outcome is unknown"), "{v}");
ui.join().unwrap();
}
}
+44 -1
View File
@@ -28,12 +28,34 @@ pub struct ControlRequest {
pub method: String,
pub params: Value,
pub reply: Sender<ControlResponse>,
/// Who has the request: [`PENDING`], [`TAKEN`] (the app runs it) or [`ABANDONED`] (the
/// sender gave up on it, so the app must not run it late: issue #719).
pub state: std::sync::Arc<std::sync::atomic::AtomicU8>,
}
/// Queued, not yet run.
pub const PENDING: u8 = 0;
/// The app has started it: it will answer.
pub const TAKEN: u8 = 1;
/// Its sender gave up waiting (the reply timeout expired): it is dropped unrun.
pub const ABANDONED: u8 = 2;
impl ControlRequest {
pub fn new(method: impl Into<String>, params: Value) -> (Self, std::sync::mpsc::Receiver<ControlResponse>) {
let (tx, rx) = std::sync::mpsc::channel();
(Self { method: method.into(), params, reply: tx }, rx)
(Self { method: method.into(), params, reply: tx, state: Default::default() }, rx)
}
/// Claim the request for running; false when its sender already gave up on it.
pub fn take(&self) -> bool {
use std::sync::atomic::Ordering;
self.state.compare_exchange(PENDING, TAKEN, Ordering::SeqCst, Ordering::SeqCst).is_ok() || self.state.load(Ordering::SeqCst) == TAKEN
}
/// Give up on the request; false when the app has already started it (it will answer).
pub fn abandon(state: &std::sync::atomic::AtomicU8) -> bool {
use std::sync::atomic::Ordering;
state.compare_exchange(PENDING, ABANDONED, Ordering::SeqCst, Ordering::SeqCst).is_ok() || state.load(Ordering::SeqCst) == ABANDONED
}
}
@@ -402,3 +424,24 @@ pub fn save_screenshot(app: &mut SlideApp, image: &egui::ColorImage, path: Optio
None => json!({"ok": false, "error": "no writer configured"}),
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used, clippy::expect_used)]
use super::*;
/// `take` and `abandon` are exclusive: whoever wins, the request runs at most once (#719).
#[test]
fn take_and_abandon_are_exclusive() {
let (req, _rx) = ControlRequest::new("m", json!({}));
assert!(req.take(), "a pending request is taken");
assert!(!ControlRequest::abandon(&req.state), "a taken request is not abandoned");
assert!(req.take(), "a taken request stays taken");
let (req, _rx) = ControlRequest::new("m", json!({}));
assert!(ControlRequest::abandon(&req.state), "a pending request is abandoned");
assert!(!req.take(), "an abandoned request is never taken");
assert!(ControlRequest::abandon(&req.state), "abandoning twice stays abandoned");
}
}
+4
View File
@@ -1879,6 +1879,10 @@ impl SlideApp {
fn drain_control(&mut self, ctx: &egui::Context) {
let Some(rx) = self.control_rx.take() else { return };
while let Ok(req) = rx.try_recv() {
// A request whose sender gave up waiting (reply timeout) must not run late (#719).
if !req.take() {
continue;
}
let reply = req.reply.clone();
match control::handle(self, ctx, &req) {
control::Outcome::Done(v) => {
+3 -1
View File
@@ -6,7 +6,9 @@ lines: one request object per line, one reply per line.
The server bounds what a client can hold: 16 connections at once (more get `{"ok":false,"error":"busy: …"}`
and are closed), 16 MiB per request line (longer closes the connection), 64 queued requests (more are
answered `busy`), 10 minutes idle and 30 seconds per blocked reply write before the socket is closed, and
60 seconds waiting for the UI (`"error":"timeout"`).
60 seconds waiting for the UI (`"error":"timeout"`). A request still queued when that wait expires is
abandoned: the UI skips it, so a retry cannot run it twice. If the UI had already started it, the reply
says the outcome is unknown instead of `timeout`.
```json
{"id": 1, "method": "engine.execute", "params": {"command": "shape.insert", "params": {"preset": "roundRect", "rect": [100, 100, 240, 120], "text": "Hello"}}}