Recover terminal output gaps with resume repaint

Cargo.lock

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

Cargo.toml

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

scripts/tui-harness.sh

@@ -9,6 +9,7 @@ for test_name in \
   tui_control_sequences_survive_transport_verbatim \
   unicode_output_survives_transport \
   tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution \
+  btop_style_graph_output_stays_raw_when_retransmit_history_overflows \
   large_tui_paint_is_delivered_in_mtu_safe_frames \
   resume_snapshot_preserves_alternate_screen_mode
 do

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

src/bin/dosh-server.rs

@@ -3336,7 +3336,7 @@ async fn broadcast_output(
     let sends = {
         let mut locked = state.lock().expect("server state poisoned");
         let scrollback = locked.config.scrollback;
-        let retransmit_window = locked.config.retransmit_window;
+        let retransmit_window = locked.config.retransmit_window.max(1);
         let session = locked
             .sessions
             .get_mut(&output.session)
@@ -3374,20 +3374,18 @@ async fn broadcast_output(
             client.send_seq += 1;
             // Count traffic toward the packet-count rekey trigger (spec §11).
             client.epoch_packets = client.epoch_packets.saturating_add(1);
-            // Only materialize a screen snapshot when this client has fallen too
+            // Live terminal bytes are never rewritten into vt100 snapshots. A
-            // far behind; the common case sends the raw output bytes and must not
+            // snapshot is useful for attach/resume, but substituting it during a
-            // clone the whole vt100 grid per client per packet.
+            // running TUI can lose graph/background-cell details that apps like
-            let (bytes, snapshot) = if client.pending.len() >= retransmit_window {
+            // btop rely on. If retransmit history is full, prune only history.
-                client.pending.clear();
+            while client.pending.len() >= retransmit_window {
-                (screen_snapshot(session.parser.screen()), true)
+                client.pending.pop_front();
-            } else {
+            }
-                (output.bytes.clone(), false)
-            };
             let frame = Frame {
                 session: output.session.clone(),
                 output_seq,
-                bytes,
+                bytes: output.bytes.clone(),
-                snapshot,
+                snapshot: false,
                 closed: false,
             };
             let body = protocol::to_body(&frame)?;
@@ -3400,9 +3398,6 @@ async fn broadcast_output(
                 SERVER_TO_CLIENT,
                 &body,
             )?;
-            while client.pending.len() >= retransmit_window {
-                client.pending.pop_front();
-            }
             client.pending.push_back(PendingFrame {
                 output_seq,
                 packet: packet.clone(),

tests/integration_smoke.rs

@@ -112,6 +112,21 @@ fn write_server_config_with_output_interval(
     config
 }
 
+fn write_server_config_with_retransmit_window(
+    dir: &tempfile::TempDir,
+    port: u16,
+    retransmit_window: u64,
+) -> std::path::PathBuf {
+    let config = write_server_config(dir, port);
+    let mut raw = fs::read_to_string(&config).unwrap();
+    raw = raw.replace(
+        "retransmit_window = 256\n",
+        &format!("retransmit_window = {retransmit_window}\n"),
+    );
+    fs::write(&config, raw).unwrap();
+    config
+}
+
 fn start_server(dir: &tempfile::TempDir, config: &std::path::Path) -> Child {
     let server = env!("CARGO_BIN_EXE_dosh-server");
     let child = Command::new(server)
@@ -1694,6 +1709,69 @@ fn tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution() {
     );
 }
 
+#[test]
+fn btop_style_graph_output_stays_raw_when_retransmit_history_overflows() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config_with_retransmit_window(&dir, port, 3);
+    let mut server = start_server(&dir, &config);
+    let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write");
+
+    let graph = "DOSH_BTOP_NET ⠀⠁⠃⠇⡇⣇⣧⣷⣿";
+    let input = Input {
+        bytes: format!(
+            "stty -echo; \
+             printf '\\033[?1049h\\033[?2026h\\033[?25l'; \
+             for i in 1 2 3 4 5 6 7 8 9 10; do \
+               printf '\\033[6;4H\\033[38;2;80;220;140m{} %s\\033[0m' \"$i\"; \
+             done; \
+             printf '\\033[?25h\\033[?2026l\\033[?1049l'\n",
+            graph
+        )
+        .into_bytes(),
+    };
+    send_encrypted(
+        &socket,
+        port,
+        PacketKind::Input,
+        ok.client_id,
+        2,
+        0,
+        &bootstrap.session_key,
+        &protocol::to_body(&input).unwrap(),
+    );
+
+    let mut text = String::new();
+    let mut snapshots = 0usize;
+    let deadline = std::time::Instant::now() + Duration::from_secs(3);
+    while std::time::Instant::now() < deadline
+        && !(text.matches("DOSH_BTOP_NET").count() >= 10 && text.contains("\x1b[?2026l"))
+    {
+        if let Some((_header, frame)) = recv_frame(&socket, &bootstrap.session_key) {
+            if frame.snapshot {
+                snapshots += 1;
+            }
+            text.push_str(&String::from_utf8_lossy(&frame.bytes));
+        }
+    }
+
+    let _ = server.kill();
+    let _ = server.wait();
+
+    assert_eq!(
+        snapshots, 0,
+        "live btop-style graph output must not be replaced by snapshots; got {snapshots}"
+    );
+    assert!(
+        text.matches("DOSH_BTOP_NET").count() >= 10
+            && text.contains("⣿")
+            && text.contains("\x1b[38;2;80;220;140m")
+            && text.contains("\x1b[?2026h")
+            && text.contains("\x1b[?2026l"),
+        "expected raw btop-style graph output to survive under retransmit pressure, got {text:?}"
+    );
+}
+
 #[test]
 fn large_tui_paint_is_delivered_in_mtu_safe_frames() {
     let dir = tempfile::tempdir().unwrap();