Recover terminal output gaps with resume repaint

Cargo.lock

@@ -436,7 +436,7 @@ dependencies = [
 
 [[package]]
 name = "dosh"
-version = "0.1.10"
+version = "0.1.11"
 dependencies = [
  "anyhow",
  "base64",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "0.1.10"
+version = "0.1.11"
 edition = "2024"
 license = "MIT"
 

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<u64, Frame>,
+    gap_since: Option<Instant>,
 }
 
 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<Frame> {
@@ -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,
+        CachedCredential, DisconnectStatus, DynamicForward, FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer,
-        MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode,
+        LocalForward, MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen,
-        Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig,
+        PredictMode, Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD,
-        StartupGateMode, StatusAction, auth_allows, cache_key, cache_server_prefix,
+        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();