diff --git a/Cargo.lock b/Cargo.lock index 85d2a61..82d7dd8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "0.1.13" +version = "0.1.14" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index a2fafea..1a43bc9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "0.1.13" +version = "0.1.14" edition = "2024" license = "MIT" diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 675126a..a43f830 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -4271,7 +4271,6 @@ async fn run_terminal( let mut status_tick = tokio::time::interval(Duration::from_secs(1)); let mut resize_tick = tokio::time::interval(Duration::from_millis(250)); let mut stream_retransmit_tick = tokio::time::interval(Duration::from_millis(200)); - let mut frame_gap_tick = tokio::time::interval(Duration::from_millis(100)); 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 @@ -4499,36 +4498,6 @@ async fn run_terminal( _ = resize_tick.tick() => { maybe_send_resize(&socket, addr, &cred, &mut send_seq, &mut last_size).await?; } - _ = frame_gap_tick.tick() => { - if frame_buffer.gap_wait_elapsed(Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS)) - && let Some(frame) = reconnect( - &socket, - &mut cred, - &mut send_seq, - last_size, - &mut frame_buffer, - &mut predictor, - ) - .await? - { - refresh_live_addr(&mut addr, &cred)?; - if !forward_only { - render_frame(&frame)?; - predictor.observe_output(&frame.bytes); - } - last_packet_at = Instant::now(); - flush_pending_user_input( - &socket, - addr, - &cred, - &mut send_seq, - &mut predictor, - &mut pending_user_input, - &mut pending_user_input_bytes, - ) - .await?; - } - } recv = socket.recv_from(&mut recv_buf) => { let (n, _) = recv?; let Ok(packet) = protocol::decode(&recv_buf[..n]) else { @@ -6592,18 +6561,14 @@ fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool { .any(|window| window == needle) } -const FRAME_GAP_RESYNC_AFTER_MS: u64 = 500; - #[derive(Default)] struct FrameBuffer { pending: BTreeMap, - gap_since: Option, } impl FrameBuffer { fn clear(&mut self) { self.pending.clear(); - self.gap_since = None; } fn accept(&mut self, frame: Frame, last_rendered_seq: &mut u64) -> Vec { @@ -6612,7 +6577,6 @@ impl FrameBuffer { return Vec::new(); } self.pending.clear(); - self.gap_since = None; *last_rendered_seq = frame.output_seq; return vec![frame]; } @@ -6626,30 +6590,8 @@ impl FrameBuffer { *last_rendered_seq = frame.output_seq; ready.push(frame); } - self.refresh_gap(*last_rendered_seq); ready } - - fn refresh_gap(&mut self, last_rendered_seq: u64) { - let waiting_on_missing = self - .pending - .first_key_value() - .is_some_and(|(&seq, _)| seq > last_rendered_seq.saturating_add(1)); - if waiting_on_missing { - self.gap_since.get_or_insert_with(Instant::now); - } else { - self.gap_since = None; - } - } - - fn gap_wait_elapsed(&self, wait: Duration) -> bool { - self.gap_since.is_some_and(|since| since.elapsed() >= wait) - } - - #[cfg(test)] - fn backdate_gap_for_test(&mut self, elapsed: Duration) { - self.gap_since = Some(Instant::now() - elapsed); - } } /// Read the current terminal size and, if it changed since `last_size`, send a @@ -7061,10 +7003,10 @@ const TERMINAL_CLEANUP: &[u8] = concat!( #[cfg(test)] mod tests { use super::{ - CachedCredential, DisconnectStatus, DynamicForward, FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, - LocalForward, MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, - PredictMode, Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, - SshConfig, StartupGateMode, StatusAction, auth_allows, cache_key, cache_server_prefix, + CachedCredential, DisconnectStatus, DynamicForward, FrameBuffer, LocalForward, + MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, 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, @@ -7483,14 +7425,12 @@ mod tests { assert!(buffer.accept(test_frame(12, false), &mut last).is_empty()); assert_eq!(last, 10); - assert!(!buffer.gap_wait_elapsed(Duration::from_secs(60))); let ready = buffer.accept(test_frame(11, false), &mut last); assert_eq!(ready.len(), 2); assert_eq!(ready[0].output_seq, 11); assert_eq!(ready[1].output_seq, 12); assert_eq!(last, 12); - assert!(!buffer.gap_wait_elapsed(Duration::from_secs(0))); } #[test] @@ -7509,28 +7449,39 @@ mod tests { let mut last = 10; assert!(buffer.accept(test_frame(12, false), &mut last).is_empty()); - buffer.backdate_gap_for_test(Duration::from_secs(10)); - assert!(buffer.gap_wait_elapsed(Duration::from_millis(1))); let ready = buffer.accept(test_frame(20, true), &mut last); assert_eq!(ready.len(), 1); assert_eq!(ready[0].output_seq, 20); assert_eq!(last, 20); - assert!(!buffer.gap_wait_elapsed(Duration::from_millis(1))); assert!(buffer.accept(test_frame(12, false), &mut last).is_empty()); } #[test] - fn frame_buffer_marks_gap_ready_for_resync_after_wait() { + fn frame_buffer_keeps_gap_buffered_without_snapshot_resync() { let mut buffer = FrameBuffer::default(); let mut last = 10; assert!(buffer.accept(test_frame(13, false), &mut last).is_empty()); assert_eq!(last, 10); - assert!(!buffer.gap_wait_elapsed(Duration::from_secs(60))); + assert!(buffer.accept(test_frame(14, false), &mut last).is_empty()); + assert_eq!(last, 10); - buffer.backdate_gap_for_test(Duration::from_secs(1)); - assert!(buffer.gap_wait_elapsed(Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS))); + let ready = buffer.accept(test_frame(11, false), &mut last); + assert_eq!(ready.len(), 1); + assert_eq!(ready[0].output_seq, 11); + assert_eq!(last, 11); + + let ready = buffer.accept(test_frame(12, false), &mut last); + assert_eq!(ready.len(), 3); + assert_eq!( + ready + .iter() + .map(|frame| frame.output_seq) + .collect::>(), + vec![12, 13, 14] + ); + assert_eq!(last, 14); } #[test] diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index e7aa86b..aa7a057 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -3519,11 +3519,9 @@ async fn retransmit_pending( if pending.output_seq <= client.last_acked { continue; } - if now.duration_since(pending.last_sent) >= Duration::from_millis(200) - && pending.attempts < 8 - { + if now.duration_since(pending.last_sent) >= Duration::from_millis(200) { pending.last_sent = now; - pending.attempts += 1; + pending.attempts = pending.attempts.saturating_add(1); sends.push((client.endpoint, pending.packet.clone())); } } diff --git a/tests/integration_smoke.rs b/tests/integration_smoke.rs index 4ba8e72..f17a640 100644 --- a/tests/integration_smoke.rs +++ b/tests/integration_smoke.rs @@ -1392,18 +1392,10 @@ fn server_retransmits_unacked_output_frames() { ); let mut seen = Vec::new(); - let mut duplicate = None; - let deadline = std::time::Instant::now() + Duration::from_secs(3); + let deadline = std::time::Instant::now() + Duration::from_secs(4); while std::time::Instant::now() < deadline { if let Some((header, frame)) = recv_frame(&socket, &bootstrap.session_key) { if String::from_utf8_lossy(&frame.bytes).contains("DOSH_RETRANSMIT") { - if seen - .iter() - .any(|(_, output_seq)| *output_seq == frame.output_seq) - { - duplicate = Some((header.seq, frame.output_seq)); - break; - } seen.push((header.seq, frame.output_seq)); } } @@ -1412,9 +1404,18 @@ fn server_retransmits_unacked_output_frames() { let _ = server.kill(); let _ = server.wait(); + let max_same_output_seq = seen + .iter() + .map(|(_, output_seq)| { + seen.iter() + .filter(|(_, candidate)| candidate == output_seq) + .count() + }) + .max() + .unwrap_or(0); assert!( - duplicate.is_some(), - "expected retransmitted same output seq, seen={seen:?}" + max_same_output_seq >= 10, + "expected retransmission to continue past the old 8-attempt ceiling, seen={seen:?}" ); }