From 5ce738cf5397d0d8a50ed1aa6335297073cc0f0f Mon Sep 17 00:00:00 2001 From: Ogulcan Celik Date: Sun, 24 May 2026 18:03:24 +0300 Subject: fix: avoid app event queue stalls refs #265 --- src/app/api.rs | 9 ++++++- src/app/mod.rs | 69 ++++++++++++++++++++++++++++++++++++++++++++--- src/app/runtime.rs | 17 +++++++++++- src/server/headless.rs | 72 +++++++++++++++++++++++++++++++++++++++++++++++--- 4 files changed, 157 insertions(+), 10 deletions(-) (limited to 'src') diff --git a/src/app/api.rs b/src/app/api.rs index a867f4d7..6f88f6c3 100644 --- a/src/app/api.rs +++ b/src/app/api.rs @@ -322,7 +322,14 @@ impl App { } pub(crate) fn handle_api_request(&mut self, request: crate::api::schema::Request) -> String { - self.drain_internal_events(); + self.drain_all_internal_events(); + self.handle_api_request_after_internal_events_drained(request) + } + + pub(crate) fn handle_api_request_after_internal_events_drained( + &mut self, + request: crate::api::schema::Request, + ) -> String { use crate::api::schema::{Method, ResponseResult, SuccessResponse}; let response = match request.method { diff --git a/src/app/mod.rs b/src/app/mod.rs index 8a981226..bacbf0ce 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -93,6 +93,9 @@ pub struct App { pub(crate) config_reloaded_from_disk: bool, } +pub(crate) const APP_EVENT_CHANNEL_CAPACITY: usize = 256; +pub(crate) const APP_EVENT_DRAIN_LIMIT: usize = 64; + pub(crate) enum LoopEvent { Timer, Internal(AppEvent), @@ -221,7 +224,7 @@ impl App { ) -> Self { let (prefix_code, prefix_mods) = config.prefix_key(); crate::kitty_graphics::set_enabled(config.experimental.kitty_graphics); - let (event_tx, event_rx) = mpsc::channel::(64); + let (event_tx, event_rx) = mpsc::channel::(APP_EVENT_CHANNEL_CAPACITY); let render_notify = Arc::new(Notify::new()); let render_dirty = Arc::new(AtomicBool::new(false)); @@ -561,7 +564,8 @@ impl App { needs_render = true; } - // Drain internal events first so API reads observe fresh pane state. + // Drain a bounded internal-event batch for responsiveness. API handlers + // perform an exhaustive drain before reading pane/runtime state. if self.drain_internal_events() { needs_render = true; } @@ -1353,6 +1357,57 @@ mod tests { assert!(app.render_dirty.load(Ordering::Acquire)); } + #[test] + fn internal_event_drain_limits_work_per_tick() { + let mut app = test_app(); + for i in 0..=APP_EVENT_DRAIN_LIMIT { + app.event_tx + .try_send(AppEvent::UpdateReady { + version: format!("2.0.{i}"), + install_command: "herdr install".into(), + }) + .unwrap(); + } + + assert!(app.drain_internal_events()); + + let expected_version = format!("2.0.{}", APP_EVENT_DRAIN_LIMIT - 1); + assert_eq!( + app.state.update_available.as_deref(), + Some(expected_version.as_str()) + ); + assert!(app.event_rx.try_recv().is_ok()); + } + + #[test] + fn api_request_drains_all_pending_internal_events_before_reading_state() { + let mut app = test_app(); + for i in 0..=APP_EVENT_DRAIN_LIMIT { + app.event_tx + .try_send(AppEvent::UpdateReady { + version: format!("3.0.{i}"), + install_command: "herdr install".into(), + }) + .unwrap(); + } + + let response = app.handle_api_request(crate::api::schema::Request { + id: "req_server_stop_after_events".into(), + method: crate::api::schema::Method::ServerStop( + crate::api::schema::EmptyParams::default(), + ), + }); + let response: serde_json::Value = serde_json::from_str(&response).unwrap(); + + assert_eq!(response["result"]["type"], "ok"); + let expected_version = format!("3.0.{APP_EVENT_DRAIN_LIMIT}"); + assert_eq!( + app.state.update_available.as_deref(), + Some(expected_version.as_str()) + ); + assert!(app.event_rx.try_recv().is_err()); + } + #[test] fn startup_uses_configured_agent_panel_scope() { let mut config = Config::default(); @@ -2631,7 +2686,7 @@ mod tests { AgentState::Working ); - for i in 0..64 { + for i in 0..APP_EVENT_CHANNEL_CAPACITY { app.event_tx .try_send(AppEvent::UpdateReady { version: format!("9.9.{i}"), @@ -2667,7 +2722,13 @@ mod tests { .expect("state change should enqueue once queue space is available") .expect("app event receiver should still be alive"); - app.drain_internal_events(); + let max_drains = (APP_EVENT_CHANNEL_CAPACITY / APP_EVENT_DRAIN_LIMIT) + 2; + for _ in 0..max_drains { + if app.state.terminals.get(&terminal_id).unwrap().state == AgentState::Idle { + break; + } + app.drain_internal_events(); + } assert_eq!( app.state.terminals.get(&terminal_id).unwrap().state, diff --git a/src/app/runtime.rs b/src/app/runtime.rs index 7f7306cc..284fb882 100644 --- a/src/app/runtime.rs +++ b/src/app/runtime.rs @@ -418,8 +418,23 @@ impl App { } pub(crate) fn drain_internal_events(&mut self) -> bool { + self.drain_internal_events_up_to(super::APP_EVENT_DRAIN_LIMIT) + } + + pub(crate) fn drain_all_internal_events(&mut self) -> bool { + let mut had_event = false; + while self.drain_internal_events_up_to(super::APP_EVENT_DRAIN_LIMIT) { + had_event = true; + } + had_event + } + + fn drain_internal_events_up_to(&mut self, limit: usize) -> bool { let mut had_event = false; - while let Ok(ev) = self.event_rx.try_recv() { + for _ in 0..limit { + let Ok(ev) = self.event_rx.try_recv() else { + break; + }; had_event = true; self.handle_internal_event(ev); } diff --git a/src/server/headless.rs b/src/server/headless.rs index 632f205d..a6d3f6a5 100644 --- a/src/server/headless.rs +++ b/src/server/headless.rs @@ -213,7 +213,8 @@ impl HeadlessServer { needs_render = true; } - // 2. Drain internal events. + // 2. Drain a bounded internal-event batch. API handlers perform an + // exhaustive forwarding-aware drain before reading pane/runtime state. if self.drain_internal_events_with_forwarding() { needs_render = true; } @@ -974,13 +975,36 @@ impl HeadlessServer { /// - Detect when a toast is set on AppState and forward as /// `ServerMessage::Notify` to the foreground client for terminal/system delivery. fn drain_internal_events_with_forwarding(&mut self) -> bool { + self.drain_internal_events_with_forwarding_up_to(crate::app::APP_EVENT_DRAIN_LIMIT) + .1 + } + + fn drain_all_internal_events_with_forwarding(&mut self) -> bool { let mut changed = false; - while let Ok(ev) = self.app.event_rx.try_recv() { - changed |= self.handle_internal_event_with_forwarding(ev); + loop { + let (had_event, batch_changed) = + self.drain_internal_events_with_forwarding_up_to(crate::app::APP_EVENT_DRAIN_LIMIT); + changed |= batch_changed; + if !had_event { + break; + } } changed } + fn drain_internal_events_with_forwarding_up_to(&mut self, limit: usize) -> (bool, bool) { + let mut had_event = false; + let mut changed = false; + for _ in 0..limit { + let Ok(ev) = self.app.event_rx.try_recv() else { + break; + }; + had_event = true; + changed |= self.handle_internal_event_with_forwarding(ev); + } + (had_event, changed) + } + fn drain_client_sound_config_reload_request(&mut self) { if !self.app.state.request_client_sound_config_reload { return; @@ -1440,6 +1464,7 @@ impl HeadlessServer { } let changed = api::request_changes_ui(&msg.request); + let changed = self.drain_all_internal_events_with_forwarding() || changed; // Capture toast and effective pane states before the API call so we can // forward resulting client-local notifications. API requests like @@ -1490,7 +1515,8 @@ impl HeadlessServer { .unwrap_or_else(|_| "{}".to_string()) }) } else { - self.app.handle_api_request(msg.request) + self.app + .handle_api_request_after_internal_events_drained(msg.request) }; let _ = msg.respond_to.send(response); @@ -2241,6 +2267,44 @@ mod tests { } } + #[test] + fn headless_api_request_drains_all_pending_internal_events_before_reading_state() { + let mut server = test_headless_server(); + for i in 0..=crate::app::APP_EVENT_DRAIN_LIMIT { + server + .app + .event_tx + .try_send(AppEvent::UpdateReady { + version: format!("4.0.{i}"), + install_command: "herdr install".into(), + }) + .unwrap(); + } + + let (respond_to, response_rx) = std::sync::mpsc::channel(); + assert!( + server.handle_api_request_with_shutdown_check(api::ApiRequestMessage { + request: api::schema::Request { + id: "headless_stop_after_events".into(), + method: api::schema::Method::ServerStop(api::schema::EmptyParams::default()), + }, + respond_to, + }) + ); + let response = response_rx + .recv_timeout(Duration::from_millis(100)) + .unwrap(); + let response: serde_json::Value = serde_json::from_str(&response).unwrap(); + + assert_eq!(response["result"]["type"], "ok"); + let expected_version = format!("4.0.{}", crate::app::APP_EVENT_DRAIN_LIMIT); + assert_eq!( + server.app.state.update_available.as_deref(), + Some(expected_version.as_str()) + ); + assert!(server.app.event_rx.try_recv().is_err()); + } + fn test_client_writer() -> ( ClientWriter, std::sync::mpsc::Receiver>, -- cgit v1.2.3