From 5c54601cdff0d3416d460fd6db633005ef01ecf4 Mon Sep 17 00:00:00 2001 From: DuProcess <273172371+DuProcess@users.noreply.github.com> Date: Sat, 4 Jul 2026 16:58:15 -0400 Subject: [PATCH] Prepare Dosh 1.0.0 rc1 --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/bin/dosh-client.rs | 191 ++++++++++++++++++++++++++++++++------- tests/hostile_network.rs | 6 +- 4 files changed, 165 insertions(+), 36 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 028de54..e2796ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "0.1.17" +version = "1.0.0-rc1" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index af493e7..5f707a5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "0.1.17" +version = "1.0.0-rc1" edition = "2024" license = "MIT" diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 3e47aaf..f0cc564 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -3524,6 +3524,60 @@ fn resolve_addr(host: &str, port: u16) -> Result { .ok_or_else(|| anyhow!("no UDP address resolved for {host}:{port}")) } +async fn send_terminal_udp(socket: &UdpSocket, packet: &[u8], addr: SocketAddr) -> Result { + match socket.send_to(packet, addr).await { + Ok(_) => Ok(true), + Err(err) if is_transient_udp_send_error(&err) => Ok(false), + Err(err) => Err(err).with_context(|| format!("send UDP packet to {addr}")), + } +} + +fn is_transient_udp_send_error(err: &std::io::Error) -> bool { + matches!( + err.kind(), + std::io::ErrorKind::Interrupted + | std::io::ErrorKind::TimedOut + | std::io::ErrorKind::WouldBlock + ) || err.raw_os_error().is_some_and(is_transient_udp_os_error) +} + +#[cfg(unix)] +fn is_transient_udp_os_error(code: i32) -> bool { + matches!( + code, + libc::EADDRNOTAVAIL + | libc::ECONNREFUSED + | libc::ECONNRESET + | libc::EHOSTDOWN + | libc::EHOSTUNREACH + | libc::ENETDOWN + | libc::ENETRESET + | libc::ENETUNREACH + | libc::ETIMEDOUT + ) +} + +#[cfg(windows)] +fn is_transient_udp_os_error(code: i32) -> bool { + matches!( + code, + 10049 // WSAEADDRNOTAVAIL + | 10050 // WSAENETDOWN + | 10051 // WSAENETUNREACH + | 10052 // WSAENETRESET + | 10054 // WSAECONNRESET + | 10060 // WSAETIMEDOUT + | 10061 // WSAECONNREFUSED + | 10064 // WSAEHOSTDOWN + | 10065 // WSAEHOSTUNREACH + ) +} + +#[cfg(not(any(unix, windows)))] +fn is_transient_udp_os_error(_code: i32) -> bool { + false +} + #[allow(clippy::too_many_arguments)] async fn try_native_auth( socket: &UdpSocket, @@ -4337,7 +4391,13 @@ async fn run_terminal( && cred.mode != "view-only" && !forward_only { - send_input(&socket, addr, &cred, &mut send_seq, bytes).await?; + if !send_input(&socket, addr, &cred, &mut send_seq, bytes.clone()).await? { + queue_stale_pending_user_input( + &mut pending_user_input, + &mut pending_user_input_bytes, + bytes, + )?; + } startup_input_hold_until = Some(Instant::now() + STARTUP_INPUT_HOLD); startup_gate_mode = StartupGateMode::HoldAll; } @@ -4411,8 +4471,15 @@ async fn run_terminal( && let Some((send_now, mut hold_for_later)) = split_after_command_submit(&bytes) { - predictor.observe_input(&send_now)?; - send_input(&socket, addr, &cred, &mut send_seq, send_now).await?; + if send_input(&socket, addr, &cred, &mut send_seq, send_now.clone()).await? { + predictor.observe_input(&send_now)?; + } else { + queue_stale_pending_user_input( + &mut pending_user_input, + &mut pending_user_input_bytes, + send_now, + )?; + } startup_input_hold_until = Some( Instant::now() + post_submit_hold_duration(&hold_for_later), @@ -4431,15 +4498,23 @@ async fn run_terminal( )?; continue; } else if !hold_for_later.is_empty() { - predictor.observe_input(&hold_for_later)?; - send_input( + let sent = send_input( &socket, addr, &cred, &mut send_seq, - std::mem::take(&mut hold_for_later), + hold_for_later.clone(), ) .await?; + if sent { + predictor.observe_input(&hold_for_later)?; + } else { + queue_stale_pending_user_input( + &mut pending_user_input, + &mut pending_user_input_bytes, + std::mem::take(&mut hold_for_later), + )?; + } } continue; } @@ -4477,8 +4552,15 @@ async fn run_terminal( .await?; } } else { - predictor.observe_input(&bytes)?; - send_input(&socket, addr, &cred, &mut send_seq, bytes).await?; + if send_input(&socket, addr, &cred, &mut send_seq, bytes.clone()).await? { + predictor.observe_input(&bytes)?; + } else { + queue_stale_pending_user_input( + &mut pending_user_input, + &mut pending_user_input_bytes, + bytes, + )?; + } } } } @@ -4667,7 +4749,7 @@ async fn run_terminal( CLIENT_TO_SERVER, b"", )?; - socket.send_to(&ack, addr).await?; + let _ = send_terminal_udp(&socket, &ack, addr).await?; } PacketKind::AttachReject => { let reject: AttachReject = protocol::from_body(&packet.body)?; @@ -5102,7 +5184,7 @@ async fn run_terminal( b"", )?; send_seq += 1; - socket.send_to(&packet, addr).await?; + let _ = send_terminal_udp(&socket, &packet, addr).await?; } // Re-evaluate the prediction display policy on every timer tick so // a latency spike (or recovery) flips speculation on/off promptly @@ -5416,6 +5498,23 @@ fn queue_pending_user_input_with_filter( Ok(()) } +fn requeue_pending_user_input_front( + pending: &mut VecDeque, + pending_bytes: &mut usize, + input: PendingUserInput, +) -> Result<()> { + let next = pending_bytes + .checked_add(input.bytes.len()) + .ok_or_else(|| anyhow!("pending input buffer overflow"))?; + anyhow::ensure!( + next <= MAX_PENDING_USER_INPUT_BYTES, + "pending input buffer full while disconnected" + ); + *pending_bytes = next; + pending.push_front(input); + Ok(()) +} + #[derive(Debug, Clone, PartialEq, Eq)] struct PendingUserInput { bytes: Vec, @@ -5486,8 +5585,19 @@ async fn flush_pending_user_input( if bytes.is_empty() { continue; } - predictor.observe_input(&bytes)?; - send_input(socket, addr, cred, send_seq, bytes).await?; + if send_input(socket, addr, cred, send_seq, bytes.clone()).await? { + predictor.observe_input(&bytes)?; + } else { + requeue_pending_user_input_front( + pending, + pending_bytes, + PendingUserInput { + bytes, + strip_mouse_reports: false, + }, + )?; + break; + } } Ok(()) } @@ -5619,7 +5729,7 @@ async fn send_input( cred: &CachedCredential, send_seq: &mut u64, bytes: Vec, -) -> Result<()> { +) -> Result { let body = protocol::to_body(&Input { bytes })?; let packet = protocol::encode_encrypted( PacketKind::Input, @@ -5631,8 +5741,7 @@ async fn send_input( &body, )?; *send_seq += 1; - socket.send_to(&packet, addr).await?; - Ok(()) + send_terminal_udp(socket, &packet, addr).await } async fn send_stream_open( @@ -5975,7 +6084,7 @@ async fn send_stream_packet( body, )?; *send_seq += 1; - socket.send_to(&packet, addr).await?; + let _ = send_terminal_udp(socket, &packet, addr).await?; Ok(()) } @@ -6739,7 +6848,7 @@ async fn send_resize( &body, )?; *send_seq += 1; - socket.send_to(&packet, addr).await?; + let _ = send_terminal_udp(socket, &packet, addr).await?; Ok(()) } @@ -6759,7 +6868,7 @@ async fn send_ack( b"", )?; *send_seq += 1; - socket.send_to(&packet, addr).await?; + let _ = send_terminal_udp(socket, &packet, addr).await?; Ok(()) } @@ -6774,7 +6883,7 @@ async fn detach_once(socket: &UdpSocket, cred: &CachedCredential, seq: u64) -> R CLIENT_TO_SERVER, b"", )?; - socket.send_to(&packet, addr).await?; + let _ = send_terminal_udp(socket, &packet, addr).await?; Ok(()) } @@ -7112,18 +7221,19 @@ mod tests { PredictMode, Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig, StartupGateMode, StatusAction, auth_allows, cache_key, cache_server_prefix, clear_cached_credentials, ensure_tui_safe_status_overlay, input_matches_escape, - is_local_status_target, is_resume_response_for_client, latest_release_download_url, - load_first_native_identity_with_prompt, parse_dynamic_forward, parse_escape_key, - parse_local_forward, parse_remote_forward, parse_ssh_config, post_submit_hold_duration, - queue_pending_user_input, raw_contains_host_table, recv_response_until, refresh_live_addr, - release_tag_download_url, release_tag_from_effective_url, release_version_from_tag, - render_status_clear, render_status_overlay, requested_env, resolved_startup_command, - retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host, - server_version_mismatch, should_flush_terminal_input_after_contact, - 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, - strip_stale_mouse_reports, toml_bare_key_or_quoted, update_check_requested, - update_version_status, upsert_managed_block, valid_forward_host, vscode_safe_alias, + is_local_status_target, is_resume_response_for_client, is_transient_udp_os_error, + latest_release_download_url, load_first_native_identity_with_prompt, parse_dynamic_forward, + parse_escape_key, parse_local_forward, parse_remote_forward, parse_ssh_config, + post_submit_hold_duration, queue_pending_user_input, raw_contains_host_table, + recv_response_until, refresh_live_addr, release_tag_download_url, + release_tag_from_effective_url, release_version_from_tag, render_status_clear, + render_status_overlay, requested_env, resolved_startup_command, retransmit_stream_opens, + rewrite_forward_command, selected_predict_mode, selected_udp_host, server_version_mismatch, + should_flush_terminal_input_after_contact, 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, strip_stale_mouse_reports, + toml_bare_key_or_quoted, update_check_requested, update_version_status, + upsert_managed_block, valid_forward_host, vscode_safe_alias, }; use dosh::config::{ClientConfig, CommandExtension, HostConfig}; use dosh::native::EnvVar; @@ -8353,6 +8463,25 @@ mod tests { assert_eq!(pending_bytes, MAX_PENDING_USER_INPUT_BYTES); } + #[test] + fn transient_udp_send_errors_are_not_terminal_fatal() { + #[cfg(unix)] + { + assert!(is_transient_udp_os_error(libc::ENETUNREACH)); + assert!(is_transient_udp_os_error(libc::EHOSTUNREACH)); + assert!(is_transient_udp_os_error(libc::EADDRNOTAVAIL)); + assert!(!is_transient_udp_os_error(libc::EINVAL)); + } + + #[cfg(windows)] + { + assert!(is_transient_udp_os_error(10051)); // WSAENETUNREACH + assert!(is_transient_udp_os_error(10065)); // WSAEHOSTUNREACH + assert!(is_transient_udp_os_error(10049)); // WSAEADDRNOTAVAIL + assert!(!is_transient_udp_os_error(10022)); // WSAEINVAL + } + } + #[test] fn stale_mouse_reports_are_stripped_from_pending_input() { let input = b"\x1b[<35;152;1Mhello\x1b[<0;107;10m"; diff --git a/tests/hostile_network.rs b/tests/hostile_network.rs index 7e122c5..10a714b 100644 --- a/tests/hostile_network.rs +++ b/tests/hostile_network.rs @@ -886,13 +886,13 @@ fn bad_network_tui_work_soak_30m() { let mut seq = 2u64; let mut iteration = 0u64; while Instant::now() < deadline { - if iteration % 2 == 0 { + if iteration.is_multiple_of(2) { relay.arm_reorder(); } - if iteration > 0 && iteration % 7 == 0 { + if iteration > 0 && iteration.is_multiple_of(7) { let _ = relay.rebind_upstream(); } - if iteration > 0 && iteration % 11 == 0 { + if iteration > 0 && iteration.is_multiple_of(11) { relay.set_drop_s2c(100); thread::sleep(Duration::from_millis(750)); relay.set_drop_s2c(20);