Compare commits

...

2 Commits

Author SHA1 Message Date
DuProcess e11ba1a652 Bump version to 1.0.0-rc16
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-10 22:49:14 -04:00
DuProcess 07c4b321d1 Harden persistent screen mirroring
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-10 22:48:32 -04:00
3 changed files with 36 additions and 9 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]]
name = "dosh"
version = "1.0.0-rc15"
version = "1.0.0-rc16"
dependencies = [
"anyhow",
"base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dosh"
version = "1.0.0-rc15"
version = "1.0.0-rc16"
edition = "2024"
license = "MIT"
+34 -7
View File
@@ -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<std::path::PathBuf>) -> 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<Mutex<ServerState>>) {
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<Mutex<ServerState>>) {
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<Mutex<ServerState>>, 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<Mutex<ServerState>>) {
// 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());