From 2568c55afac04d13d748019925b16889f56450a9 Mon Sep 17 00:00:00 2001 From: midasmatterhorn <248507632+midasmatterhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:37:58 +0300 Subject: [PATCH] control: a request that timed out while queued is abandoned, never run late (#719) --- apps/deckcraft/src/control_server.rs | 54 +++++++++++++++++++++++++++- crates/ui-egui/src/control.rs | 45 ++++++++++++++++++++++- crates/ui-egui/src/lib.rs | 4 +++ docs/control-protocol.md | 4 ++- 4 files changed, 104 insertions(+), 3 deletions(-) diff --git a/apps/deckcraft/src/control_server.rs b/apps/deckcraft/src/control_server.rs index 966657b0..0abac12a 100644 --- a/apps/deckcraft/src/control_server.rs +++ b/apps/deckcraft/src/control_server.rs @@ -151,10 +151,23 @@ fn serve(stream: TcpStream, tx: &SyncSender, 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(); + } } diff --git a/crates/ui-egui/src/control.rs b/crates/ui-egui/src/control.rs index 8f474485..8292321c 100644 --- a/crates/ui-egui/src/control.rs +++ b/crates/ui-egui/src/control.rs @@ -28,12 +28,34 @@ pub struct ControlRequest { pub method: String, pub params: Value, pub reply: Sender, + /// 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, } +/// 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, params: Value) -> (Self, std::sync::mpsc::Receiver) { 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"); + } +} diff --git a/crates/ui-egui/src/lib.rs b/crates/ui-egui/src/lib.rs index f020f5f5..e2864b9f 100644 --- a/crates/ui-egui/src/lib.rs +++ b/crates/ui-egui/src/lib.rs @@ -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) => { diff --git a/docs/control-protocol.md b/docs/control-protocol.md index b7b6f013..0bd20302 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -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"}}}