Drop stale mouse reports on reconnect

Cargo.lock

@@ -436,7 +436,7 @@ dependencies = [
 
 [[package]]
 name = "dosh"
-version = "0.1.4"
+version = "0.1.5"
 dependencies = [
  "anyhow",
  "base64",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "0.1.4"
+version = "0.1.5"
 edition = "2024"
 license = "MIT"
 

src/bin/dosh-client.rs

@@ -2862,7 +2862,7 @@ async fn run_terminal(
     let mut stream_sent_data: HashMap<u64, BTreeMap<u64, PendingStreamChunk>> = HashMap::new();
     let mut stream_next_recv_offset: HashMap<u64, u64> = HashMap::new();
     let mut stream_recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>> = HashMap::new();
-    let mut pending_user_input: VecDeque<Vec<u8>> = VecDeque::new();
+    let mut pending_user_input: VecDeque<PendingUserInput> = VecDeque::new();
     let mut pending_user_input_bytes = 0usize;
     let mut startup_input_hold_until: Option<Instant> = None;
     let mut startup_gate_mode = StartupGateMode::HoldControl;
@@ -2988,7 +2988,7 @@ async fn run_terminal(
                                 continue;
                             }
                             if last_packet_at.elapsed() >= Duration::from_secs(2) {
-                                queue_pending_user_input(
+                                queue_stale_pending_user_input(
                                     &mut pending_user_input,
                                     &mut pending_user_input_bytes,
                                     bytes,
@@ -3878,9 +3878,26 @@ fn refresh_live_addr(addr: &mut SocketAddr, cred: &CachedCredential) -> Result<(
 }
 
 fn queue_pending_user_input(
-    pending: &mut VecDeque<Vec<u8>>,
+    pending: &mut VecDeque<PendingUserInput>,
     pending_bytes: &mut usize,
     bytes: Vec<u8>,
+) -> Result<()> {
+    queue_pending_user_input_with_filter(pending, pending_bytes, bytes, false)
+}
+
+fn queue_stale_pending_user_input(
+    pending: &mut VecDeque<PendingUserInput>,
+    pending_bytes: &mut usize,
+    bytes: Vec<u8>,
+) -> Result<()> {
+    queue_pending_user_input_with_filter(pending, pending_bytes, bytes, true)
+}
+
+fn queue_pending_user_input_with_filter(
+    pending: &mut VecDeque<PendingUserInput>,
+    pending_bytes: &mut usize,
+    bytes: Vec<u8>,
+    strip_mouse_reports: bool,
 ) -> Result<()> {
     let next = pending_bytes
         .checked_add(bytes.len())
@@ -3890,10 +3907,19 @@ fn queue_pending_user_input(
         "pending input buffer full while disconnected"
     );
     *pending_bytes = next;
-    pending.push_back(bytes);
+    pending.push_back(PendingUserInput {
+        bytes,
+        strip_mouse_reports,
+    });
     Ok(())
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct PendingUserInput {
+    bytes: Vec<u8>,
+    strip_mouse_reports: bool,
+}
+
 fn split_after_command_submit(bytes: &[u8]) -> Option<(Vec<u8>, Vec<u8>)> {
     let split_at = bytes
         .iter()
@@ -3933,24 +3959,126 @@ async fn flush_pending_user_input(
     cred: &CachedCredential,
     send_seq: &mut u64,
     predictor: &mut Predictor,
-    pending: &mut VecDeque<Vec<u8>>,
+    pending: &mut VecDeque<PendingUserInput>,
     pending_bytes: &mut usize,
 ) -> Result<()> {
-    while let Some(bytes) = pending.pop_front() {
+    while let Some(input) = pending.pop_front() {
+        let PendingUserInput {
+            bytes,
+            strip_mouse_reports,
+        } = input;
         *pending_bytes = pending_bytes.saturating_sub(bytes.len());
+        let bytes = if strip_mouse_reports {
+            strip_stale_mouse_reports(&bytes)
+        } else {
+            bytes
+        };
+        if bytes.is_empty() {
+            continue;
+        }
         predictor.observe_input(&bytes)?;
         send_input(socket, addr, cred, send_seq, bytes).await?;
     }
     Ok(())
 }
 
+fn strip_stale_mouse_reports(bytes: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(bytes.len());
+    let mut offset = 0;
+    while offset < bytes.len() {
+        if let Some(len) = stale_mouse_report_len(bytes, offset) {
+            offset += len;
+            continue;
+        }
+        if stale_mouse_report_maybe_incomplete(bytes, offset) {
+            break;
+        }
+        out.push(bytes[offset]);
+        offset += 1;
+    }
+    out
+}
+
+fn stale_mouse_report_len(bytes: &[u8], offset: usize) -> Option<usize> {
+    match bytes.get(offset).copied()? {
+        0x1b if bytes.get(offset + 1) == Some(&b'[') => match bytes.get(offset + 2).copied() {
+            Some(b'M') if offset + 6 <= bytes.len() => Some(6),
+            Some(b'<') => mouse_report_params_len(bytes, offset + 3, 2).map(|len| len + 3),
+            Some(byte) if byte.is_ascii_digit() => {
+                mouse_report_params_len(bytes, offset + 2, 2).map(|len| len + 2)
+            }
+            _ => None,
+        },
+        0x9b => match bytes.get(offset + 1).copied() {
+            Some(b'M') if offset + 5 <= bytes.len() => Some(5),
+            Some(b'<') => mouse_report_params_len(bytes, offset + 2, 2).map(|len| len + 2),
+            Some(byte) if byte.is_ascii_digit() => {
+                mouse_report_params_len(bytes, offset + 1, 2).map(|len| len + 1)
+            }
+            _ => None,
+        },
+        byte if byte.is_ascii_digit() => mouse_report_params_len(bytes, offset, 1),
+        _ => None,
+    }
+}
+
+fn mouse_report_params_len(bytes: &[u8], offset: usize, min_semicolons: usize) -> Option<usize> {
+    let mut semicolons = 0usize;
+    let mut cursor = offset;
+    while let Some(byte) = bytes.get(cursor).copied() {
+        match byte {
+            b'0'..=b'9' => cursor += 1,
+            b';' => {
+                semicolons += 1;
+                cursor += 1;
+            }
+            b'M' | b'm' if semicolons >= min_semicolons => return Some(cursor + 1 - offset),
+            _ => return None,
+        }
+    }
+    None
+}
+
+fn stale_mouse_report_maybe_incomplete(bytes: &[u8], offset: usize) -> bool {
+    let Some(rest) = bytes.get(offset..) else {
+        return false;
+    };
+    match rest {
+        [0x1b] | [0x1b, b'['] | [0x1b, b'[', b'<'] => true,
+        [0x1b, b'[', b'M', ..] if rest.len() < 6 => true,
+        [0x1b, b'[', b'<', params @ ..] | [0x1b, b'[', params @ ..]
+            if params_are_mouse_prefix(params) =>
+        {
+            true
+        }
+        [0x9b] | [0x9b, b'<'] => true,
+        [0x9b, b'M', ..] if rest.len() < 5 => true,
+        [0x9b, b'<', params @ ..] | [0x9b, params @ ..] if params_are_mouse_prefix(params) => true,
+        params if params_are_mouse_prefix(params) => true,
+        _ => false,
+    }
+}
+
+fn params_are_mouse_prefix(params: &[u8]) -> bool {
+    let mut semicolons = 0usize;
+    let mut saw_digit = false;
+    for byte in params {
+        match byte {
+            b'0'..=b'9' => saw_digit = true,
+            b';' if saw_digit => semicolons += 1,
+            _ => return false,
+        }
+    }
+    saw_digit && semicolons >= 1
+}
+
 async fn flush_startup_input_if_ready(
     socket: &UdpSocket,
     addr: SocketAddr,
     cred: &CachedCredential,
     send_seq: &mut u64,
     predictor: &mut Predictor,
-    pending: (&mut VecDeque<Vec<u8>>, &mut usize),
+    pending: (&mut VecDeque<PendingUserInput>, &mut usize),
     hold: (&mut Option<Instant>, &mut StartupGateMode),
 ) -> Result<()> {
     let (startup_hold_until, startup_gate_mode) = hold;
@@ -5433,8 +5561,8 @@ mod tests {
         retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
         server_version_mismatch, should_hold_during_startup_gate, should_hold_post_submit_input,
         split_after_command_submit, ssh_destination_host, ssh_username, ssh_with_user,
-        startup_command, status_ssh_target, toml_bare_key_or_quoted, update_check_requested,
-        valid_forward_host,
+        startup_command, status_ssh_target, strip_stale_mouse_reports, toml_bare_key_or_quoted,
+        update_check_requested, valid_forward_host,
     };
     use dosh::config::{ClientConfig, CommandExtension, HostConfig};
     use dosh::native::EnvVar;
@@ -6516,8 +6644,8 @@ mod tests {
         queue_pending_user_input(&mut pending, &mut pending_bytes, b"abc".to_vec()).unwrap();
         queue_pending_user_input(&mut pending, &mut pending_bytes, b"def".to_vec()).unwrap();
         assert_eq!(pending_bytes, 6);
-        assert_eq!(pending.pop_front().unwrap(), b"abc");
-        assert_eq!(pending.pop_front().unwrap(), b"def");
+        assert_eq!(pending.pop_front().unwrap().bytes, b"abc");
+        assert_eq!(pending.pop_front().unwrap().bytes, b"def");
 
         let mut pending = VecDeque::new();
         let mut pending_bytes = MAX_PENDING_USER_INPUT_BYTES;
@@ -6526,6 +6654,28 @@ mod tests {
         assert_eq!(pending_bytes, MAX_PENDING_USER_INPUT_BYTES);
     }
 
+    #[test]
+    fn stale_mouse_reports_are_stripped_from_pending_input() {
+        let input = b"\x1b[<35;152;1Mhello\x1b[<0;107;10m";
+        assert_eq!(strip_stale_mouse_reports(input), b"hello");
+    }
+
+    #[test]
+    fn stale_mouse_tail_reports_are_stripped_from_pending_input() {
+        let input = b"35;152;1M35;149;1Mls\r";
+        assert_eq!(strip_stale_mouse_reports(input), b"ls\r");
+        assert_eq!(strip_stale_mouse_reports(b"152;1Mls\r"), b"ls\r");
+    }
+
+    #[test]
+    fn non_mouse_escape_input_survives_stale_filter() {
+        assert_eq!(strip_stale_mouse_reports(b"\x1b[A"), b"\x1b[A");
+        assert_eq!(
+            strip_stale_mouse_reports(b"\x1b[200~paste\x1b[201~"),
+            b"\x1b[200~paste\x1b[201~"
+        );
+    }
+
     #[test]
     fn escape_key_parser_accepts_mosh_style_controls() {
         assert_eq!(parse_escape_key("^]").unwrap(), Some(vec![0x1d]));