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");