From 07c4b321d1c949428b6e25c4c8c7a9dded92a9eb Mon Sep 17 00:00:00 2001 From: DuProcess <273172371+DuProcess@users.noreply.github.com> Date: Fri, 10 Jul 2026 22:48:32 -0400 Subject: [PATCH] Harden persistent screen mirroring --- src/bin/dosh-server.rs | 41 ++++++++++++++++++++++++++++++++++------- 1 file changed, 34 insertions(+), 7 deletions(-) diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 4c6721b..740f588 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -53,6 +53,7 @@ const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024; /// have accumulated since the last mirror. Keeps the atomic write off the /// per-packet hot path while bounding how much fresh output a crash can lose. const SCREEN_PERSIST_BYTES: usize = 4096; +const SCREEN_PERSIST_MAX_AGE: Duration = Duration::from_millis(250); const IMPLICIT_EMPTY_SESSION_GRACE_SECS: u64 = 300; /// Sentinel `target_host` sent to the client on a server-initiated `StreamOpen` @@ -251,7 +252,7 @@ async fn serve(config_path: Option) -> Result<()> { if config.persist_sessions { let flush_state = Arc::clone(&state); tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_secs(2)); + let mut interval = tokio::time::interval(SCREEN_PERSIST_MAX_AGE); loop { interval.tick().await; flush_persistent_screens(&flush_state); @@ -376,6 +377,10 @@ struct Session { /// `output_seq` at the last successful screen mirror, so the periodic flush /// skips sessions whose screen has not changed. last_persisted_seq: u64, + /// Last time this session's screen was considered for a mirror. Combined + /// with the byte threshold so low-throughput prompt/TUI updates persist + /// quickly without writing on every packet in high-throughput bursts. + last_screen_persist_at: Instant, } #[derive(Clone)] @@ -511,6 +516,7 @@ impl ServerState { persistent, bytes_since_persist: 0, last_persisted_seq: 0, + last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE, }, ); Ok(()) @@ -648,6 +654,7 @@ impl ServerState { persistent: true, bytes_since_persist: 0, last_persisted_seq: output_seq, + last_screen_persist_at: Instant::now(), }, ); eprintln!( @@ -3451,16 +3458,20 @@ async fn broadcast_output( session.recent.pop_front(); } // For a persistent session, mirror the screen to disk periodically so a - // post-restart reattach repaints. Throttled by accumulated bytes to keep - // the (atomic) write off the per-packet hot path. The actual file write - // happens after the lock is dropped. + // post-restart reattach repaints. Throttled by accumulated bytes or a + // short wall-clock age so low-throughput prompt/TUI changes persist + // quickly while large bursts still avoid per-packet disk writes. The + // actual file write happens after the lock is dropped. if session.persistent { session.bytes_since_persist = session .bytes_since_persist .saturating_add(output.bytes.len()); - if session.bytes_since_persist >= SCREEN_PERSIST_BYTES { + let now = Instant::now(); + if session.bytes_since_persist >= SCREEN_PERSIST_BYTES + || now.duration_since(session.last_screen_persist_at) >= SCREEN_PERSIST_MAX_AGE + { session.bytes_since_persist = 0; - session.last_persisted_seq = output_seq; + session.last_screen_persist_at = now; let screen = session.parser.screen(); persist_screen = Some(( output.session.clone(), @@ -3519,6 +3530,8 @@ async fn broadcast_output( persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot) { eprintln!("screen persist for {name} failed: {err:#}"); + } else { + mark_screen_persisted(state, &name, output_seq); } } for (endpoint, packet) in sends { @@ -3800,8 +3813,8 @@ fn flush_persistent_screens(state: &Arc>) { if !session.persistent || session.output_seq == session.last_persisted_seq { continue; } - session.last_persisted_seq = session.output_seq; session.bytes_since_persist = 0; + session.last_screen_persist_at = Instant::now(); let screen = session.parser.screen(); to_write.push(( name.clone(), @@ -3818,10 +3831,19 @@ fn flush_persistent_screens(state: &Arc>) { persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot) { eprintln!("screen flush for {name} failed: {err:#}"); + } else { + mark_screen_persisted(state, &name, output_seq); } } } +fn mark_screen_persisted(state: &Arc>, name: &str, output_seq: u64) { + let mut locked = state.lock().expect("server state poisoned"); + if let Some(session) = locked.sessions.get_mut(name) { + session.last_persisted_seq = session.last_persisted_seq.max(output_seq); + } +} + fn cleanup_disconnected_clients(state: &Arc>) { // Sessions removed here are dropped after the lock is released so their // shells are killed (PtyHandle::drop) without holding up the event loop. @@ -4382,6 +4404,7 @@ mod tests { persistent: false, bytes_since_persist: 0, last_persisted_seq: 0, + last_screen_persist_at: Instant::now(), }, ); state.client_index.insert(client_id, "test".to_string()); @@ -4445,6 +4468,7 @@ mod tests { persistent: false, bytes_since_persist: 0, last_persisted_seq: 0, + last_screen_persist_at: Instant::now(), }, ); state.client_index.insert(client_id, "test".to_string()); @@ -4520,6 +4544,7 @@ mod tests { persistent: false, bytes_since_persist: 0, last_persisted_seq: 0, + last_screen_persist_at: Instant::now(), }, ); state.client_index.insert(client_id, "test".to_string()); @@ -4689,6 +4714,7 @@ mod tests { persistent: false, bytes_since_persist: 0, last_persisted_seq: 0, + last_screen_persist_at: Instant::now(), }, ); // Keep the O(1) conn_id index in sync, matching `insert_client`. @@ -4785,6 +4811,7 @@ mod tests { persistent: false, bytes_since_persist: 0, last_persisted_seq: 0, + last_screen_persist_at: Instant::now(), }, ); state.client_index.insert(client_id, "test".to_string());