Stabilize terminal frame recovery

Cargo.lock

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

Cargo.toml

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

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<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> {
@@ -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,
+        CachedCredential, DisconnectStatus, DynamicForward, FrameBuffer, LocalForward,
-        LocalForward, MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen,
+        MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode,
-        PredictMode, Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD,
+        Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig,
-        SshConfig, StartupGateMode, StatusAction, auth_allows, cache_key, cache_server_prefix,
+        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));
+        let ready = buffer.accept(test_frame(11, false), &mut last);
-        assert!(buffer.gap_wait_elapsed(Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS)));
+        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<_>>(),
+            vec![12, 13, 14]
+        );
+        assert_eq!(last, 14);
     }
 
     #[test]

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)
+                    if now.duration_since(pending.last_sent) >= Duration::from_millis(200) {
-                        && pending.attempts < 8
-                    {
                         pending.last_sent = now;
-                        pending.attempts += 1;
+                        pending.attempts = pending.attempts.saturating_add(1);
                         sends.push((client.endpoint, pending.packet.clone()));
                     }
                 }

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(4);
-    let deadline = std::time::Instant::now() + Duration::from_secs(3);
     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(),
+        max_same_output_seq >= 10,
-        "expected retransmitted same output seq, seen={seen:?}"
+        "expected retransmission to continue past the old 8-attempt ceiling, seen={seen:?}"
     );
 }