diff --git a/Cargo.lock b/Cargo.lock index 82d7dd8..91e2f37 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "0.1.14" +version = "0.1.15" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index 1a43bc9..aebaf15 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "0.1.14" +version = "0.1.15" edition = "2024" license = "MIT" diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index a43f830..8855114 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -4271,6 +4271,7 @@ 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(250)); 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 @@ -4498,6 +4499,36 @@ 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.resync_due() + && 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 { @@ -6561,14 +6592,22 @@ fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool { .any(|window| window == needle) } +const FRAME_GAP_RESYNC_AFTER_MS: u64 = 2_500; +const FRAME_GAP_RESYNC_COOLDOWN_MS: u64 = 5_000; +const FRAME_GAP_PENDING_RESYNC_THRESHOLD: usize = 128; + #[derive(Default)] struct FrameBuffer { pending: BTreeMap, + gap_since: Option, + last_resync_at: Option, } impl FrameBuffer { fn clear(&mut self) { self.pending.clear(); + self.gap_since = None; + self.last_resync_at = None; } fn accept(&mut self, frame: Frame, last_rendered_seq: &mut u64) -> Vec { @@ -6577,6 +6616,8 @@ impl FrameBuffer { return Vec::new(); } self.pending.clear(); + self.gap_since = None; + self.last_resync_at = None; *last_rendered_seq = frame.output_seq; return vec![frame]; } @@ -6590,8 +6631,47 @@ 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; + self.last_resync_at = None; + } + } + + fn resync_due(&mut self) -> bool { + let Some(gap_since) = self.gap_since else { + return false; + }; + let now = Instant::now(); + let waited = now.duration_since(gap_since); + if waited < Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS) + && self.pending.len() < FRAME_GAP_PENDING_RESYNC_THRESHOLD + { + return false; + } + if self.last_resync_at.is_some_and(|last| { + now.duration_since(last) < Duration::from_millis(FRAME_GAP_RESYNC_COOLDOWN_MS) + }) { + return false; + } + self.last_resync_at = Some(now); + true + } + + #[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 @@ -7003,10 +7083,10 @@ const TERMINAL_CLEANUP: &[u8] = concat!( #[cfg(test)] mod tests { use super::{ - 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, + 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, 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, @@ -7466,6 +7546,7 @@ mod tests { assert_eq!(last, 10); assert!(buffer.accept(test_frame(14, false), &mut last).is_empty()); assert_eq!(last, 10); + assert!(!buffer.resync_due()); let ready = buffer.accept(test_frame(11, false), &mut last); assert_eq!(ready.len(), 1); @@ -7484,6 +7565,26 @@ mod tests { assert_eq!(last, 14); } + #[test] + fn frame_buffer_resyncs_old_gap_with_cooldown() { + let mut buffer = FrameBuffer::default(); + let mut last = 10; + + assert!(buffer.accept(test_frame(13, false), &mut last).is_empty()); + buffer.backdate_gap_for_test(Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS + 10)); + assert!(buffer.resync_due()); + assert!( + !buffer.resync_due(), + "resync must be rate-limited so snapshots cannot churn the TUI" + ); + + 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.resync_due()); + } + #[test] fn frame_buffer_ignores_stale_snapshot() { let mut buffer = FrameBuffer::default();