diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 731599c..f9b191b 100644 --- a/src/bin/dosh-client.rs +++ b/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::>(); + 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>, mpsc::Receiver>) { + 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::>(); + 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>> { )) } -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 ---