aboutsummaryrefslogtreecommitdiff
path: root/src/app
diff options
context:
space:
mode:
authorOgulcan Celik <ogulcancelik@gmail.com>2026-05-24 18:03:24 +0300
committerOgulcan Celik <ogulcancelik@gmail.com>2026-05-24 18:03:24 +0300
commit5ce738cf5397d0d8a50ed1aa6335297073cc0f0f (patch)
treebed3b4f720c58f76562709d22ce30aabf734eaf2 /src/app
parent26205a2ac5e21053bd94a0925efcfca44c6f549b (diff)
fix: avoid app event queue stalls
refs #265
Diffstat (limited to 'src/app')
-rw-r--r--src/app/api.rs9
-rw-r--r--src/app/mod.rs69
-rw-r--r--src/app/runtime.rs17
3 files changed, 89 insertions, 6 deletions
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::<AppEvent>(64);
+ let (event_tx, event_rx) = mpsc::channel::<AppEvent>(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;
}
@@ -1354,6 +1358,57 @@ mod tests {
}
#[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();
config.ui.agent_panel_scope = crate::config::AgentPanelScopeConfig::Current;
@@ -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);
}