mirror of
https://github.com/storytold/gridcraft.git
synced 2026-10-24 10:46:50 -06:00
Merge PR #1971: Control channel: a request its sender gave up on is not run later
This commit is contained in:
@@ -119,14 +119,23 @@ fn serve(stream: TcpStream, tx: Sender<ControlRequest>, ctx: egui::Context) {
|
||||
Line::Reject(e) => e,
|
||||
Line::Request { id, method, params } => {
|
||||
let (req, rx) = ControlRequest::new(method, params);
|
||||
let state = std::sync::Arc::clone(&req.state);
|
||||
if tx.send(req).is_err() {
|
||||
break;
|
||||
}
|
||||
// Wake the UI thread: egui only runs frames when something happens.
|
||||
ctx.request_repaint();
|
||||
let mut reply = rx
|
||||
.recv_timeout(Duration::from_secs(60))
|
||||
.unwrap_or_else(|_| json!({"ok": false, "error": "timed out waiting for the UI (is the window minimised?)"}));
|
||||
let mut reply = rx.recv_timeout(Duration::from_secs(60)).unwrap_or_else(|_| {
|
||||
// Give up on the request as well as on its reply. While it is still
|
||||
// queued this keeps it from running late -- the client was told it had
|
||||
// failed, and an agent that retried after the error ran it twice (#1914);
|
||||
// if the app had already started it, its outcome is unknown instead.
|
||||
if ControlRequest::abandon(&state) {
|
||||
json!({"ok": false, "error": "timed out waiting for the UI (is the window minimised?)"})
|
||||
} else {
|
||||
json!({"ok": false, "error": "timed out waiting for the UI, which had already started the request, so it may still finish; its outcome is unknown"})
|
||||
}
|
||||
});
|
||||
if let Some(o) = reply.as_object_mut() {
|
||||
o.insert("id".into(), id);
|
||||
}
|
||||
|
||||
@@ -29,12 +29,35 @@ pub struct ControlRequest {
|
||||
pub method: String,
|
||||
pub params: Json,
|
||||
pub reply: Sender<ControlResponse>,
|
||||
/// Who has the request: [`PENDING`], [`TAKEN`] (the app runs it) or [`ABANDONED`] (the sender
|
||||
/// gave up waiting, so the app must not run it late: a client told its request timed out had
|
||||
/// it applied anyway on the next UI pass, and an agent that retried ran it twice (#1914).
|
||||
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: Json) -> (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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -414,6 +437,11 @@ pub fn handle(app: &mut SheetApp, ctx: &egui::Context, method: &str, p: &Json) -
|
||||
pub fn poll(app: &mut SheetApp, ctx: &egui::Context) {
|
||||
let Some(rx) = app.control_rx.take() else { return };
|
||||
while let Ok(req) = rx.try_recv() {
|
||||
// A request whose sender stopped waiting is dropped unrun: running it would apply a
|
||||
// command the client was told had failed, and its reply would go nowhere (#1914).
|
||||
if !req.take() {
|
||||
continue;
|
||||
}
|
||||
let outcome = handle(app, ctx, &req.method, &req.params);
|
||||
match outcome {
|
||||
Outcome::Done(v) => {
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
//! A control request whose sender stopped waiting is dropped unrun (#1914).
|
||||
|
||||
use gridcraft_engine::Session;
|
||||
use gridcraft_engine::core::CellRef;
|
||||
use gridcraft_ui_egui::{ControlRequest, SheetApp};
|
||||
use serde_json::json;
|
||||
|
||||
fn app() -> (SheetApp, egui::Context) {
|
||||
let mut s = Session::new();
|
||||
s.new_workbook();
|
||||
(SheetApp::new(s, Default::default()), egui::Context::default())
|
||||
}
|
||||
|
||||
fn a1(app: &SheetApp) -> String {
|
||||
let d = app.session.active().unwrap();
|
||||
d.wb.active().unwrap().cell(CellRef::parse("A1").unwrap()).map(|c| format!("{:?}", c.value)).unwrap_or_else(|| "empty".into())
|
||||
}
|
||||
|
||||
/// `set` writes A1, so a request that did run is visible in the workbook.
|
||||
fn set_a1(text: &str) -> serde_json::Value {
|
||||
json!({"command": "cell.set", "params": {"cell": "A1", "input": text}})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_request_its_sender_gave_up_on_does_not_run_later() {
|
||||
let (mut app, ctx) = app();
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
app.control_rx = Some(rx);
|
||||
|
||||
// One request the sender is still waiting for, and one it gave up on after its reply timed
|
||||
// out. Before the fix both ran on the next UI pass: the client was told its command had
|
||||
// failed and the command was applied anyway, so an agent that retried ran it twice.
|
||||
let (kept, kept_rx) = ControlRequest::new("engine.execute", set_a1("kept"));
|
||||
tx.send(kept).unwrap();
|
||||
gridcraft_ui_egui::control::poll(&mut app, &ctx);
|
||||
assert!(kept_rx.try_recv().is_ok(), "a request still waited for is answered");
|
||||
assert_eq!(a1(&app), "Text(\"kept\")");
|
||||
|
||||
let (dropped, dropped_rx) = ControlRequest::new("engine.execute", set_a1("gave up"));
|
||||
assert!(ControlRequest::abandon(&dropped.state), "the sender gives up while it is queued");
|
||||
tx.send(dropped).unwrap();
|
||||
gridcraft_ui_egui::control::poll(&mut app, &ctx);
|
||||
assert_eq!(a1(&app), "Text(\"kept\")", "the abandoned request did not run");
|
||||
assert!(dropped_rx.try_recv().is_err(), "and nothing was replied to it");
|
||||
}
|
||||
|
||||
/// Which of the two sides wins the race is settled once: a request the app has started is not
|
||||
/// abandoned (the app will answer, so the sender reports an unknown outcome instead of a failure),
|
||||
/// and one already abandoned is never started.
|
||||
#[test]
|
||||
fn taking_and_abandoning_a_request_settle_once() {
|
||||
let (taken, _rx) = ControlRequest::new("engine.execute", json!({}));
|
||||
assert!(taken.take(), "a pending request can be taken");
|
||||
assert!(taken.take(), "and taking it again is the same answer");
|
||||
assert!(!ControlRequest::abandon(&taken.state), "it cannot then be abandoned");
|
||||
|
||||
let (given_up, _rx) = ControlRequest::new("engine.execute", json!({}));
|
||||
assert!(ControlRequest::abandon(&given_up.state), "a pending request can be abandoned");
|
||||
assert!(ControlRequest::abandon(&given_up.state), "and again is the same answer");
|
||||
assert!(!given_up.take(), "it cannot then be taken");
|
||||
}
|
||||
Reference in New Issue
Block a user