Prevent terminal input from starving reconnects

src/bin/dosh-client.rs

@@ -71,6 +71,7 @@ const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
 const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 const MAX_PENDING_USER_INPUT_BYTES: usize = 1024 * 1024;
+const STDIN_QUEUE_CAPACITY: usize = 64;
 const STARTUP_INPUT_HOLD: Duration = Duration::from_millis(750);
 const POST_SUBMIT_ALL_INPUT_HOLD: Duration = Duration::from_millis(120);
 const STALE_TERMINAL_INPUT_AFTER: Duration = Duration::from_secs(2);
@@ -111,6 +112,14 @@ fn terminal_size() -> (u16, u16) {
     }
 }
 
+fn terminal_resize_poll_interval_for(os: &str) -> Duration {
+    if os == "windows" {
+        Duration::from_millis(50)
+    } else {
+        Duration::from_millis(250)
+    }
+}
+
 #[derive(Debug, Clone, Parser)]
 #[command(
     name = "dosh-client",
@@ -2480,7 +2489,7 @@ async fn run_proxy_stdio_command(config: &dosh::config::ClientConfig, args: &Arg
 }
 
 async fn proxy_stdio_loop(mut transport: DoshTransport, stream_id: u64) -> Result<()> {
-    let (stdin_tx, mut stdin_rx) = mpsc::unbounded_channel::<Vec<u8>>();
+    let (stdin_tx, mut stdin_rx) = terminal_input_channel();
     std::thread::Builder::new()
         .name("dosh-proxy-stdin".to_string())
         .spawn(move || {
@@ -2490,7 +2499,7 @@ async fn proxy_stdio_loop(mut transport: DoshTransport, stream_id: u64) -> Resul
                 match stdin.read(&mut buf) {
                     Ok(0) => break,
                     Ok(n) => {
-                        if stdin_tx.send(buf[..n].to_vec()).is_err() {
+                        if stdin_tx.blocking_send(buf[..n].to_vec()).is_err() {
                             break;
                         }
                     }
@@ -6468,6 +6477,10 @@ where
     }
 }
 
+fn terminal_input_channel() -> (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>) {
+    mpsc::channel(STDIN_QUEUE_CAPACITY)
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn run_terminal(
     socket: UdpSocket,
@@ -6496,15 +6509,24 @@ async fn run_terminal(
     let mut send_seq = 2u64;
     let mut last_packet_at = Instant::now();
     let mut status_tick = tokio::time::interval(Duration::from_secs(1));
-    let mut resize_tick = tokio::time::interval(Duration::from_millis(250));
+    let mut resize_tick =
+        tokio::time::interval(terminal_resize_poll_interval_for(std::env::consts::OS));
     let mut stream_retransmit_tick =
         tokio::time::interval(dosh::transport::ADAPTIVE_RETRANSMIT_MIN);
     let mut frame_gap_tick = tokio::time::interval(Duration::from_millis(250));
+    for interval in [
+        &mut status_tick,
+        &mut resize_tick,
+        &mut stream_retransmit_tick,
+        &mut frame_gap_tick,
+    ] {
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    }
     let mut last_size = terminal_size();
     // React to terminal resize the instant it happens via SIGWINCH (mosh-style),
-    // instead of waiting up to one `resize_tick`. The 250ms poll below stays as a
-    // fallback for environments where the signal doesn't fire. `signal()` can fail
-    // (rare), in which case we rely on the poll.
+    // instead of waiting for `resize_tick`. The platform poll stays as a fallback
+    // where the signal does not fire; Windows polls faster because it has no
+    // SIGWINCH path. `signal()` can fail, in which case Unix also relies on polling.
     #[cfg(unix)]
     let mut winch =
         tokio::signal::unix::signal(tokio::signal::unix::SignalKind::window_change()).ok();
@@ -6592,7 +6614,7 @@ async fn run_terminal(
         startup_gate_mode = StartupGateMode::HoldAll;
     }
 
-    let (stdin_tx, mut stdin_rx) = mpsc::unbounded_channel::<Vec<u8>>();
+    let (stdin_tx, mut stdin_rx) = terminal_input_channel();
     let _stdin_keepalive = if forward_only {
         Some(stdin_tx)
     } else {
@@ -6605,7 +6627,9 @@ async fn run_terminal(
                     match stdin.read(&mut buf) {
                         Ok(0) => break,
                         Ok(n) => {
-                            let _ = stdin_tx.send(buf[..n].to_vec());
+                            if stdin_tx.blocking_send(buf[..n].to_vec()).is_err() {
+                                break;
+                            }
                         }
                         Err(_) => break,
                     }
@@ -6618,9 +6642,12 @@ async fn run_terminal(
     // pre-rekey frame still decrypts instead of triggering a needless reconnect.
     let mut previous_session_key: Option<[u8; 32]> = None;
     let mut recv_buf = vec![0u8; 65535];
+    let mut detach_requested = false;
     loop {
+        if detach_requested {
+            break;
+        }
         tokio::select! {
-            biased;
             stdin_msg = stdin_rx.recv() => {
                 match stdin_msg {
                     Some(mut bytes) => {
@@ -6628,9 +6655,15 @@ async fn run_terminal(
                             "client.stdin",
                             &[("bytes", dosh::trace::bytes_summary(&bytes))],
                         );
-                        if input_matches_escape(&bytes, escape_key.as_deref()) {
+                        let (prefix_len, escape_found) =
+                            input_prefix_before_escape(&bytes, escape_key.as_deref());
+                        if escape_found {
                             dosh::trace::event("client.escape", &[]);
-                            break;
+                            bytes.truncate(prefix_len);
+                            detach_requested = true;
+                            if bytes.is_empty() {
+                                break;
+                            }
                         }
                         let input_status_tick_gap = last_status_tick_at.elapsed();
                         let mut refreshed_before_input = false;
@@ -10205,11 +10238,20 @@ fn parse_escape_key(raw: &str) -> Result<Option<Vec<u8>>> {
     ))
 }
 
-fn input_matches_escape(bytes: &[u8], escape_key: Option<&[u8]>) -> bool {
+fn input_prefix_before_escape(bytes: &[u8], escape_key: Option<&[u8]>) -> (usize, bool) {
     let Some(escape_key) = escape_key else {
-        return false;
+        return (bytes.len(), false);
     };
-    !escape_key.is_empty() && bytes.windows(escape_key.len()).any(|w| w == escape_key)
+    if escape_key.is_empty() {
+        return (bytes.len(), false);
+    }
+    match bytes
+        .windows(escape_key.len())
+        .position(|window| window == escape_key)
+    {
+        Some(index) => (index, true),
+        None => (bytes.len(), false),
+    }
 }
 
 /// Resolve whether the snapshot-restored disconnect status line is enabled.
@@ -10906,23 +10948,23 @@ mod tests {
         NativeIdentityContext, POST_RECONNECT_STALE_INPUT_GRACE, POST_SUBMIT_ALL_INPUT_HOLD,
         PendingStreamControl, PendingStreamOpen, PendingWindowAdjust, PredictMode, Predictor,
         RESTART_STATUS_SCRIPT, RemoteForward, STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD,
-        STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS, STREAM_INITIAL_WINDOW, SshConfig,
-        SshPathTokenContext, StartupGateMode, StatusAction, TERMINAL_CLEANUP,
+        STDIN_QUEUE_CAPACITY, STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS, STREAM_INITIAL_WINDOW,
+        SshConfig, SshPathTokenContext, StartupGateMode, StatusAction, TERMINAL_CLEANUP,
         TERMINAL_SNAPSHOT_RESET, UpdateOptions, UpdateRole, auth_allows, cache_key,
         cache_server_prefix, cleanup_stream_state, clear_cached_credentials,
         ensure_tui_safe_status_overlay, expand_ssh_path_tokens, first_resolved_addr,
-        imported_host_block, input_contains_focus_in, input_matches_escape, is_local_status_target,
-        is_resume_response_for_client, load_first_native_identity_with_prompt,
-        local_symlink_target_is_dir, local_username_from_env, native_proxy_udp_warning,
-        newest_client_trace_path_from, note_authenticated_contact, note_snapshot_rendered,
-        parse_dynamic_forward, parse_escape_key, parse_local_forward, parse_remote_forward,
-        parse_single_remote_path, parse_ssh_config, parse_trace_line, parse_trace_options,
-        parse_trace_report_options, parse_trace_summary, parse_update_options,
-        post_submit_hold_duration, queue_or_send_stream_data, queue_pending_user_input,
-        queue_stale_pending_user_input, raw_contains_host_table, recv_response_until,
-        refresh_live_addr, release_artifact_name_for, release_tag_download_url,
-        release_version_from_tag, remote_unix_server_update_script, render_frame_bytes,
-        render_status_overlay, requested_env, resolve_forward_agent_endpoint,
+        imported_host_block, input_contains_focus_in, input_prefix_before_escape,
+        is_local_status_target, is_resume_response_for_client,
+        load_first_native_identity_with_prompt, local_symlink_target_is_dir,
+        local_username_from_env, native_proxy_udp_warning, newest_client_trace_path_from,
+        note_authenticated_contact, note_snapshot_rendered, parse_dynamic_forward,
+        parse_escape_key, parse_local_forward, parse_remote_forward, parse_single_remote_path,
+        parse_ssh_config, parse_trace_line, parse_trace_options, parse_trace_report_options,
+        parse_trace_summary, parse_update_options, post_submit_hold_duration,
+        queue_or_send_stream_data, queue_pending_user_input, queue_stale_pending_user_input,
+        raw_contains_host_table, recv_response_until, refresh_live_addr, release_artifact_name_for,
+        release_tag_download_url, release_version_from_tag, remote_unix_server_update_script,
+        render_frame_bytes, render_status_overlay, requested_env, resolve_forward_agent_endpoint,
         resolved_startup_command, retire_stream_state, retransmit_stream_closes,
         retransmit_stream_eofs, retransmit_stream_opens, retransmit_stream_window_adjusts,
         rewrite_forward_command, sanitize_trace_name, selected_predict_mode, selected_udp_host,
@@ -10934,9 +10976,10 @@ mod tests {
         ssh_config_uses_proxy, ssh_config_word_for_os, ssh_destination_host, ssh_username,
         ssh_with_user, startup_command, status_ssh_target, strip_stale_mouse_reports,
         strip_terminal_focus_reports, strip_unowned_terminal_reports, summarize_trace_file,
-        summarize_trace_file_with_mode, terminal_private_mode_transition, toml_bare_key_or_quoted,
-        top_trace_events, trace_report_warnings, unix_update_script, update_installer_url,
-        update_version_status, upsert_managed_block, valid_forward_host, vscode_command_candidates,
+        summarize_trace_file_with_mode, terminal_input_channel, terminal_private_mode_transition,
+        terminal_resize_poll_interval_for, toml_bare_key_or_quoted, top_trace_events,
+        trace_report_warnings, unix_update_script, update_installer_url, update_version_status,
+        upsert_managed_block, valid_forward_host, vscode_command_candidates,
         vscode_fallback_command, vscode_safe_alias, wake_repaint_retry_deadline,
         windows_command_word, windows_deferred_update_script, windows_mode_from_readonly,
         windows_powershell_command_candidates, windows_readonly_from_mode, windows_update_script,
@@ -14348,10 +14391,65 @@ mod tests {
     }
 
     #[test]
-    fn escape_key_matches_inside_stdin_chunk() {
-        assert!(input_matches_escape(b"abc\x1d", Some(&[0x1d])));
-        assert!(!input_matches_escape(b"abc", Some(&[0x1d])));
-        assert!(!input_matches_escape(b"abc\x1d", None));
+    fn escape_key_preserves_the_prefix_of_a_coalesced_stdin_chunk() {
+        assert_eq!(
+            input_prefix_before_escape(b"abc\x1dignored", Some(&[0x1d])),
+            (3, true)
+        );
+        assert_eq!(
+            input_prefix_before_escape(b"\x1dignored", Some(&[0x1d])),
+            (0, true)
+        );
+        assert_eq!(
+            input_prefix_before_escape(b"abc", Some(&[0x1d])),
+            (3, false)
+        );
+        assert_eq!(input_prefix_before_escape(b"abc\x1d", None), (4, false));
+        assert_eq!(
+            input_prefix_before_escape(b"abcENDignored", Some(b"END")),
+            (3, true)
+        );
+    }
+
+    #[tokio::test]
+    async fn terminal_input_queue_applies_bounded_backpressure() {
+        let (tx, mut rx) = terminal_input_channel();
+        for index in 0..STDIN_QUEUE_CAPACITY {
+            tx.try_send(vec![index as u8]).unwrap();
+        }
+        assert!(matches!(
+            tx.try_send(vec![0xff]),
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_))
+        ));
+        assert_eq!(rx.recv().await, Some(vec![0]));
+        tx.try_send(vec![0xff]).unwrap();
+    }
+
+    #[test]
+    fn terminal_event_loop_does_not_bias_continuous_input_over_network() {
+        let source = include_str!("dosh-client.rs");
+        let run_loop = source
+            .split("async fn run_terminal(")
+            .nth(1)
+            .and_then(|tail| tail.split("async fn start_local_forwards(").next())
+            .expect("run_terminal source");
+        assert!(run_loop.contains("tokio::select!"));
+        assert!(!run_loop.contains("biased;"));
+        assert!(run_loop.contains("stdin_msg = stdin_rx.recv()"));
+        assert!(run_loop.contains("recv = socket.recv_from(&mut recv_buf)"));
+        assert_eq!(run_loop.matches("MissedTickBehavior::Skip").count(), 1);
+    }
+
+    #[test]
+    fn windows_resize_polling_matches_interactive_frame_times() {
+        assert_eq!(
+            terminal_resize_poll_interval_for("windows"),
+            Duration::from_millis(50)
+        );
+        assert_eq!(
+            terminal_resize_poll_interval_for("macos"),
+            Duration::from_millis(250)
+        );
     }
 
     // --- Item 1: disconnect status line state machine ---