From d5bc8e3c648ce6e230339a4af1c5cf582addb943 Mon Sep 17 00:00:00 2001 From: DuProcess <273172371+DuProcess@users.noreply.github.com> Date: Thu, 2 Jul 2026 19:29:20 -0400 Subject: [PATCH] Recover terminal output gaps with resume repaint --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/bin/dosh-client.rs | 84 ++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 82 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7e99746..2f8a7f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "0.1.10" +version = "0.1.11" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index 4feaa1b..29ce1d4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "0.1.10" +version = "0.1.11" edition = "2024" license = "MIT" diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index cb139fd..675126a 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(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 @@ -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.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 { @@ -6561,14 +6592,18 @@ 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 { @@ -6577,6 +6612,7 @@ impl FrameBuffer { return Vec::new(); } self.pending.clear(); + self.gap_since = None; *last_rendered_seq = frame.output_seq; return vec![frame]; } @@ -6590,8 +6626,30 @@ 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 @@ -7003,10 +7061,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, @@ -7425,12 +7483,14 @@ 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] @@ -7449,14 +7509,30 @@ 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() { + 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))); + + buffer.backdate_gap_for_test(Duration::from_secs(1)); + assert!(buffer.gap_wait_elapsed(Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS))); + } + #[test] fn frame_buffer_ignores_stale_snapshot() { let mut buffer = FrameBuffer::default();