From 2e4d4bcd7f6fb25435d305521833fa70d875dd42 Mon Sep 17 00:00:00 2001 From: mofeng-git Date: Wed, 28 Oct 2026 12:41:51 +0800 Subject: [PATCH] fix(rustdesk): preserve throttled relative mouse movement Accumulate relative deltas in the existing pending move buffer and flush them on the existing timer or before button and wheel events. Preserve absolute coordinate replacement and trackpad throttling. Split relative motion after HID queue coalescing so OTG and CH9329 reports preserve bounded displacement under congestion. Add socket timer, input ordering, trackpad, and queue saturation regression tests. Adapted from the relative mouse improvements proposed in #308 (03866aa4), retaining the current connection writer architecture. --- src/hid/mod.rs | 177 +++++++++++++++++++++++++++++++- src/rustdesk/connection.rs | 203 +++++++++++++++++++++++++++++++++---- 2 files changed, 361 insertions(+), 19 deletions(-) diff --git a/src/hid/mod.rs b/src/hid/mod.rs index 5da308ed..f4bfd9ad 100644 --- a/src/hid/mod.rs +++ b/src/hid/mod.rs @@ -742,6 +742,73 @@ pub(crate) async fn test_controller(backend: Arc) -> Arc>, + changed: tokio::sync::Notify, + runtime: tokio::sync::watch::Sender<()>, + } + + impl RecordingBackend { + pub fn new() -> Arc { + Arc::new(Self { + events: parking_lot::Mutex::new(Vec::new()), + changed: tokio::sync::Notify::new(), + runtime: tokio::sync::watch::channel(()).0, + }) + } + + pub async fn wait_for_events(&self, count: usize) { + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let changed = self.changed.notified(); + if self.events.lock().len() >= count { + return; + } + changed.await; + } + }) + .await + .expect("HID worker should deliver mouse events"); + } + } + + #[async_trait::async_trait] + impl HidBackend for RecordingBackend { + async fn init(&self) -> Result<()> { + Ok(()) + } + async fn send_keyboard(&self, _: KeyboardEvent) -> Result<()> { + Ok(()) + } + async fn send_mouse(&self, event: MouseEvent) -> Result<()> { + self.events.lock().push(event); + self.changed.notify_one(); + Ok(()) + } + async fn reset(&self) -> Result<()> { + Ok(()) + } + async fn shutdown(&self) -> Result<()> { + Ok(()) + } + fn runtime_snapshot(&self) -> HidBackendRuntimeSnapshot { + HidBackendRuntimeSnapshot { + initialized: true, + online: true, + supports_absolute_mouse: true, + ..Default::default() + } + } + fn subscribe_runtime(&self) -> tokio::sync::watch::Receiver<()> { + self.runtime.subscribe() + } + } +} + async fn apply_backend_runtime_state( backend_type: &Arc>, runtime_state: &Arc>, @@ -784,7 +851,7 @@ async fn process_hid_event( futures::executor::block_on(async move { match event { QueuedHidEvent::Keyboard(ev) => backend.send_keyboard(ev).await, - QueuedHidEvent::Mouse(ev) => backend.send_mouse(ev).await, + QueuedHidEvent::Mouse(ev) => send_mouse_to_backend(backend.as_ref(), ev).await, QueuedHidEvent::Consumer(ev) => backend.send_consumer(ev).await, QueuedHidEvent::Reset => backend.reset().await, QueuedHidEvent::Confirmed(..) => unreachable!("nested confirmed HID event"), @@ -805,6 +872,34 @@ async fn process_hid_event( } } +async fn send_mouse_to_backend(backend: &dyn HidBackend, event: MouseEvent) -> Result<()> { + if event.event_type != MouseEventType::Move { + return backend.send_mouse(event).await; + } + + // Split after queue coalescing, otherwise a full queue could merge small + // reports back into a delta that OTG/CH9329 would truncate to signed 8-bit. + // Match the pending queue's bound to keep work finite for arbitrary callers. + let mut x = event.x.clamp(-32767, 32767); + let mut y = event.y.clamp(-32767, 32767); + loop { + let dx = x.clamp(-127, 127); + let dy = y.clamp(-127, 127); + backend + .send_mouse(MouseEvent { + x: dx, + y: dy, + ..event.clone() + }) + .await?; + x -= dx; + y -= dy; + if x == 0 && y == 0 { + return Ok(()); + } + } +} + fn device_for_backend_type(backend_type: &HidBackendType) -> Option { match backend_type { HidBackendType::Ch9329 { port, .. } => Some(port.clone()), @@ -850,6 +945,86 @@ fn merge_pending_move(pending: &mut Option, event: MouseEvent) { #[cfg(test)] mod queue_tests { use super::*; + + #[tokio::test] + async fn large_relative_motion_is_split_at_the_backend() { + let backend = test_support::RecordingBackend::new(); + send_mouse_to_backend(backend.as_ref(), MouseEvent::move_rel(300, -260)) + .await + .unwrap(); + let events = backend.events.lock(); + let deltas: Vec<_> = events.iter().map(|event| (event.x, event.y)).collect(); + assert_eq!(deltas, [(127, -127), (127, -127), (46, -6)]); + } + + #[tokio::test] + async fn backend_split_is_bounded_and_preserves_non_relative_events() { + let backend = test_support::RecordingBackend::new(); + send_mouse_to_backend(backend.as_ref(), MouseEvent::move_rel(i32::MAX, i32::MIN)) + .await + .unwrap(); + { + let events = backend.events.lock(); + assert_eq!(events.len(), 259); + assert_eq!(events.iter().map(|event| event.x).sum::(), 32767); + assert_eq!(events.iter().map(|event| event.y).sum::(), -32767); + } + backend.events.lock().clear(); + send_mouse_to_backend(backend.as_ref(), MouseEvent::move_abs(30000, 20000)) + .await + .unwrap(); + send_mouse_to_backend(backend.as_ref(), MouseEvent::scroll(-2)) + .await + .unwrap(); + let events = backend.events.lock(); + assert_eq!(events.len(), 2); + assert_eq!((events[0].x, events[0].y), (30000, 20000)); + assert_eq!(events[0].event_type, MouseEventType::MoveAbs); + assert_eq!(events[1].scroll, -2); + } + + #[tokio::test] + async fn congested_queue_splits_accumulated_relative_motion_without_loss() { + #[cfg(unix)] + let controller = HidController::new(HidBackendType::None, None); + #[cfg(not(unix))] + let controller = HidController::new(HidBackendType::None); + let backend = test_support::RecordingBackend::new(); + *controller.backend.write().await = Some(backend.clone()); + controller.sync_runtime_state_from_backend().await; + // Keep the worker stopped so saturation does not depend on scheduling. + for _ in 0..HID_EVENT_QUEUE_CAPACITY { + controller + .send_mouse(MouseEvent::move_rel(0, 0)) + .await + .unwrap(); + } + controller + .send_mouse(MouseEvent::move_rel(100, -20)) + .await + .unwrap(); + controller + .send_mouse(MouseEvent::move_rel(80, 30)) + .await + .unwrap(); + controller.start_event_worker().await; + controller + .send_mouse_confirmed(MouseEvent::scroll(1)) + .await + .unwrap(); + let events = backend.events.lock(); + let moves: Vec<_> = events + .iter() + .filter(|event| event.event_type == MouseEventType::Move) + .collect(); + assert!(moves + .iter() + .all(|event| event.x.abs() <= 127 && event.y.abs() <= 127)); + assert_eq!(moves.iter().map(|event| event.x).sum::(), 180); + assert_eq!(moves.iter().map(|event| event.y).sum::(), 10); + assert_eq!(events.last().unwrap().event_type, MouseEventType::Scroll); + } + #[cfg(unix)] #[tokio::test] async fn composite_otg_updates_and_rollback_preserve_hid_instance() { diff --git a/src/rustdesk/connection.rs b/src/rustdesk/connection.rs index 65f38a85..255a6362 100644 --- a/src/rustdesk/connection.rs +++ b/src/rustdesk/connection.rs @@ -43,6 +43,8 @@ const DEFAULT_SCREEN_HEIGHT: u32 = 1080; /// Default mouse event throttle interval (16ms ≈ 60Hz) const DEFAULT_MOUSE_THROTTLE_MS: u64 = 16; +/// Bound accumulated relative input so one flush cannot create an excessive HID burst. +const MAX_PENDING_RELATIVE_DELTA: i64 = 4096; /// Limit work retained for unauthenticated peers. const MAX_CONNECTIONS: usize = 8; @@ -213,7 +215,7 @@ pub struct Connection { last_caps_lock: bool, /// Whether relative mouse mode is currently active for this connection relative_mouse_active: bool, - /// Latest throttled move; flushed on the next HID interval. + /// Latest absolute position or accumulated relative movement, flushed on the next HID interval. pending_mouse_move: Option, /// Server-configured RustDesk video codec. configured_codec: VideoEncoderType, @@ -426,10 +428,7 @@ impl Connection { } _ = mouse_flush_interval.tick(), if self.pending_mouse_move.is_some() => { - if let Some(mouse_event) = self.pending_mouse_move.take() { - self.send_mouse_event_to_hid(&mouse_event).await; - self.input_throttler.mark_mouse_sent(); - } + self.flush_pending_mouse_move().await; } _ = &mut handshake_deadline, if self.state() != ConnectionState::Active => { @@ -1406,36 +1405,60 @@ impl Connection { // Check if this is a pure move event (no button/scroll) let is_pure_move = event_type == mouse_type::MOVE || is_relative_move; - // For pure move events, apply throttling - if is_pure_move && !self.input_throttler.should_send_mouse_move() { - // Coalesce moves instead of losing the final pointer position. - self.pending_mouse_move = Some(me.clone()); + if is_pure_move { + self.queue_mouse_move(me); + if self.input_throttler.should_send_mouse_move() { + self.flush_pending_mouse_move().await; + } return Ok(()); } // Preserve input ordering when a button/scroll event follows a move // that was waiting for the next HID poll interval. - if !is_pure_move { - if let Some(pending) = self.pending_mouse_move.take() { - if self.send_mouse_event_to_hid(&pending).await { - self.input_throttler.mark_mouse_sent(); - } - } - } + self.flush_pending_mouse_move().await; debug!("Mouse event: x={}, y={}, mask={}", me.x, me.y, me.mask); let attempted = self.send_mouse_event_to_hid(me).await; // Fractional trackpad packets that emit no HID report must not delay moves. // Keep the existing throttling behavior for other event types. - if self.hid.is_some() && !is_pure_move && (event_type != mouse_type::TRACKPAD || attempted) - { + if self.hid.is_some() && (event_type != mouse_type::TRACKPAD || attempted) { self.input_throttler.mark_mouse_sent(); } Ok(()) } + fn queue_mouse_move(&mut self, event: &MouseEvent) { + let mut next = event.clone(); + if event.mask & 0x07 == mouse_type::MOVE_RELATIVE { + let previous = self + .pending_mouse_move + .as_ref() + .filter(|pending| pending.mask & 0x07 == mouse_type::MOVE_RELATIVE); + let (x, y) = previous + .map(|pending| (pending.x, pending.y)) + .unwrap_or_default(); + next.x = (i64::from(x) + i64::from(event.x)) + .clamp(-MAX_PENDING_RELATIVE_DELTA, MAX_PENDING_RELATIVE_DELTA) + as i32; + next.y = (i64::from(y) + i64::from(event.y)) + .clamp(-MAX_PENDING_RELATIVE_DELTA, MAX_PENDING_RELATIVE_DELTA) + as i32; + } + // Absolute coordinates replace the previous sample. Changing modes also + // replaces it: absolute coordinates and relative deltas cannot be added. + self.pending_mouse_move = Some(next); + } + + async fn flush_pending_mouse_move(&mut self) { + if let Some(pending) = self.pending_mouse_move.take() { + if self.send_mouse_event_to_hid(&pending).await { + self.input_throttler.mark_mouse_sent(); + } + } + } + /// Returns whether at least one HID report was attempted, including failed sends. async fn send_mouse_event_to_hid(&mut self, event: &MouseEvent) -> bool { let mouse_events = @@ -2106,6 +2129,150 @@ async fn run_audio_streaming( #[cfg(test)] mod tests { use super::*; + use crate::hid::test_support::RecordingBackend; + + fn mouse_event(kind: i32, x: i32, y: i32) -> MouseEvent { + let mut event = MouseEvent::new(); + event.mask = kind; + event.x = x; + event.y = y; + event + } + + #[test] + fn pending_mouse_motion_accumulates_relative_and_replaces_absolute() { + let mut connection = connection(ConnectionMode::DirectIp); + connection.queue_mouse_move(&mouse_event(mouse_type::MOVE_RELATIVE, 100, -20)); + connection.queue_mouse_move(&mouse_event(mouse_type::MOVE_RELATIVE, 80, 30)); + let pending = connection.pending_mouse_move.as_ref().unwrap(); + assert_eq!((pending.x, pending.y), (180, 10)); + + connection.queue_mouse_move(&mouse_event(mouse_type::MOVE, 320, 240)); + connection.queue_mouse_move(&mouse_event(mouse_type::MOVE, 640, 480)); + let pending = connection.pending_mouse_move.as_ref().unwrap(); + assert_eq!((pending.x, pending.y), (640, 480)); + + connection.queue_mouse_move(&mouse_event(mouse_type::MOVE_RELATIVE, 12, -8)); + let pending = connection.pending_mouse_move.as_ref().unwrap(); + assert_eq!((pending.x, pending.y), (12, -8)); + connection.queue_mouse_move(&mouse_event(mouse_type::MOVE_RELATIVE, i32::MAX, i32::MIN)); + let pending = connection.pending_mouse_move.as_ref().unwrap(); + assert_eq!((pending.x, pending.y), (4096, -4096)); + } + + #[tokio::test] + async fn immediate_relative_move_includes_pending_delta_and_clears_it() { + let backend = RecordingBackend::new(); + let mut connection = connection(ConnectionMode::DirectIp); + connection.hid = Some(crate::hid::test_controller(backend.clone()).await); + connection.queue_mouse_move(&mouse_event(mouse_type::MOVE_RELATIVE, 30, -10)); + connection + .handle_mouse_event(&mouse_event(mouse_type::MOVE_RELATIVE, 20, 5)) + .await + .unwrap(); + assert!(connection.pending_mouse_move.is_none()); + backend.wait_for_events(1).await; + let events = backend.events.lock(); + assert_eq!(events.len(), 1); + assert_eq!((events[0].x, events[0].y), (50, -5)); + } + + #[tokio::test] + async fn pending_relative_motion_is_delivered_before_mouse_buttons() { + let backend = RecordingBackend::new(); + let mut connection = connection(ConnectionMode::DirectIp); + connection.hid = Some(crate::hid::test_controller(backend.clone()).await); + connection.input_throttler.mouse_interval = Duration::from_secs(60); + connection.input_throttler.mark_mouse_sent(); + connection + .handle_mouse_event(&mouse_event(mouse_type::MOVE_RELATIVE, 300, -260)) + .await + .unwrap(); + assert!(connection.pending_mouse_move.is_some()); + for kind in [mouse_type::DOWN, mouse_type::UP] { + connection + .handle_mouse_event(&mouse_event( + kind | (super::super::hid_adapter::mouse_button::LEFT << 3), + 0, + 0, + )) + .await + .unwrap(); + } + assert!(connection.pending_mouse_move.is_none()); + backend.wait_for_events(5).await; + let events = backend.events.lock(); + assert_eq!(events[..3].iter().map(|event| event.x).sum::(), 300); + assert_eq!(events[..3].iter().map(|event| event.y).sum::(), -260); + assert!(events[..3] + .iter() + .all(|event| event.event_type == crate::hid::MouseEventType::Move)); + assert_eq!(events[3].event_type, crate::hid::MouseEventType::Down); + assert_eq!(events[4].event_type, crate::hid::MouseEventType::Up); + } + + #[tokio::test] + async fn fractional_trackpad_input_does_not_restart_mouse_throttle() { + let backend = RecordingBackend::new(); + let mut connection = connection(ConnectionMode::DirectIp); + connection.hid = Some(crate::hid::test_controller(backend.clone()).await); + let previous = Instant::now() - Duration::from_millis(10); + connection.input_throttler.last_mouse_time = previous; + connection + .handle_mouse_event(&mouse_event(mouse_type::TRACKPAD, 0, 60)) + .await + .unwrap(); + assert_eq!(connection.input_throttler.last_mouse_time, previous); + connection + .handle_mouse_event(&mouse_event(mouse_type::TRACKPAD, 0, 60)) + .await + .unwrap(); + assert!(connection.input_throttler.last_mouse_time > previous); + backend.wait_for_events(1).await; + assert_eq!(backend.events.lock()[0].scroll, 1); + } + + #[tokio::test] + async fn connection_timer_delivers_final_relative_move_without_more_input() { + let backend = RecordingBackend::new(); + let mut connection = connection(ConnectionMode::DirectIp); + connection.hid = Some(crate::hid::test_controller(backend.clone()).await); + connection.input_throttler.mouse_interval = Duration::from_secs(60); + connection.input_throttler.mark_mouse_sent(); + let state = connection.state.clone(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let client = tokio::spawn(async move { TcpStream::connect(address).await.unwrap() }); + let (stream, peer) = listener.accept().await.unwrap(); + let mut client = client.await.unwrap(); + let server = tokio::spawn(async move { connection.handle_tcp(stream, peer).await }); + read_frame_with_limit(&mut client, MAX_UNAUTHENTICATED_PACKET_LENGTH) + .await + .unwrap(); + // Authentication is covered separately; exercise the real socket input + // loop and flush timer without starting unrelated media pipelines. + *state.write() = ConnectionState::Active; + let mut message = HbbMessage::new(); + message.union = Some(message::Union::MouseEvent(mouse_event( + mouse_type::MOVE_RELATIVE, + 10, + -5, + ))); + super::super::bytes_codec::write_frame(&mut client, &message.write_to_bytes().unwrap()) + .await + .unwrap(); + // No subsequent packet or button event is needed to deliver this move. + backend.wait_for_events(1).await; + let events = backend.events.lock().clone(); + assert_eq!(events.len(), 1); + assert_eq!((events[0].x, events[0].y), (10, -5)); + drop(client); + tokio::time::timeout(Duration::from_secs(2), server) + .await + .unwrap() + .unwrap() + .unwrap(); + } fn connection(mode: ConnectionMode) -> Connection { crypto::init().expect("crypto should initialize");