Rate limit terminal gap resync

Cargo.lock

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

Cargo.toml

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

src/bin/dosh-client.rs

@@ -4271,7 +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 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
@@ -4500,7 +4500,7 @@ async fn run_terminal(
                 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))
+                if frame_buffer.resync_due()
                     && let Some(frame) = reconnect(
                         &socket,
                         &mut cred,
@@ -6592,18 +6592,22 @@ fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
         .any(|window| window == needle)
 }
 
-const FRAME_GAP_RESYNC_AFTER_MS: u64 = 500;
+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<u64, Frame>,
     gap_since: Option<Instant>,
+    last_resync_at: Option<Instant>,
 }
 
 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<Frame> {
@@ -6613,6 +6617,7 @@ impl FrameBuffer {
             }
             self.pending.clear();
             self.gap_since = None;
+            self.last_resync_at = None;
             *last_rendered_seq = frame.output_seq;
             return vec![frame];
         }
@@ -6639,11 +6644,28 @@ impl FrameBuffer {
             self.gap_since.get_or_insert_with(Instant::now);
         } else {
             self.gap_since = None;
+            self.last_resync_at = None;
         }
     }
 
-    fn gap_wait_elapsed(&self, wait: Duration) -> bool {
-        self.gap_since.is_some_and(|since| since.elapsed() >= wait)
+    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)]
@@ -7483,14 +7505,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 +7529,60 @@ 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);
+        assert!(!buffer.resync_due());
 
-        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<_>>(),
+            vec![12, 13, 14]
+        );
+        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]

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()));
                     }
                 }

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:?}"
     );
 }