Palav/dosh
1
0
Fork 0
Code Issues Pull Requests Actions 58 Packages Projects Releases 51 Wiki Activity

Preserve raw terminal output for TUIs
ci / test (push) Has been cancelled
Details
ci / fuzz-smoke (push) Has been cancelled
Details
ci / windows-client (push) Has been cancelled
Details
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
Details
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
Details
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
Details
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
Details
ci / remote-bench (push) Has been cancelled
Details

Browse Source
This commit is contained in:
DuProcess
2026-07-02 18:34:09 -04:00
parent 691b58e531
commit a48aa15308
5 changed files with 61 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bin/dosh-server.rs
+3 -131
View File
@@ -232,19 +232,6 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
}
});
let pacing_state = Arc::clone(&state);
let pacing_socket = Arc::clone(&socket);
let pacing_interval_ms = config.output_frame_interval_ms.max(1);
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_millis(pacing_interval_ms));
loop {
interval.tick().await;
if let Err(err) = flush_paced_snapshots(&pacing_state, &pacing_socket).await {
eprintln!("paced snapshot error: {err:#}");
}
}
});
let cleanup_state = Arc::clone(&state);
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(5));
@@ -400,8 +387,6 @@ struct ClientState {
rows: u16,
last_seen: Instant,
pending: VecDeque<PendingFrame>,
last_frame_sent: Instant,
paced_snapshot_due: bool,
allowed_forwardings: Vec<ForwardingRequest>,
stream_writers: HashMap<u64, mpsc::Sender<Vec<u8>>>,
opened_streams: HashSet<u64>,
@@ -1167,8 +1152,6 @@ async fn handle_native_user_auth(
rows,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: req.auth.requested_forwardings.clone(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -1311,8 +1294,6 @@ async fn handle_bootstrap_attach(
rows: req.rows,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -1447,8 +1428,6 @@ async fn handle_ticket_attach(
rows: req.rows,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -3358,8 +3337,6 @@ async fn broadcast_output(
let mut locked = state.lock().expect("server state poisoned");
let scrollback = locked.config.scrollback;
let retransmit_window = locked.config.retransmit_window;
let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms);
let now = Instant::now();
let session = locked
.sessions
.get_mut(&output.session)
@@ -3394,15 +3371,6 @@ async fn broadcast_output(
}
let mut sends = Vec::new();
for (client_id, client) in session.clients.iter_mut() {
let terminal_control =
output.bytes.contains(&0x1b) || session.parser.screen().alternate_screen();
if !terminal_control
&& !frame_interval.is_zero()
&& now.duration_since(client.last_frame_sent) < frame_interval
{
client.paced_snapshot_due = true;
continue;
}
client.send_seq += 1;
// Count traffic toward the packet-count rekey trigger (spec §11).
client.epoch_packets = client.epoch_packets.saturating_add(1);
@@ -3441,8 +3409,6 @@ async fn broadcast_output(
last_sent: Instant::now(),
attempts: 0,
});
client.last_frame_sent = now;
client.paced_snapshot_due = false;
sends.push((client.endpoint, packet));
}
sends
@@ -3626,76 +3592,6 @@ async fn retransmit_pending(
Ok(())
}
async fn flush_paced_snapshots(
state: &Arc<Mutex<ServerState>>,
socket: &Arc<UdpSocket>,
) -> Result<()> {
let sends = {
let mut locked = state.lock().expect("server state poisoned");
let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms);
if frame_interval.is_zero() {
return Ok(());
}
let retransmit_window = locked.config.retransmit_window;
let now = Instant::now();
let mut sends = Vec::new();
for (session_name, session) in locked.sessions.iter_mut() {
let due = session.clients.values().any(|client| {
client.paced_snapshot_due
&& now.duration_since(client.last_frame_sent) >= frame_interval
});
if !due {
continue;
}
let snapshot = screen_snapshot(session.parser.screen());
let output_seq = session.output_seq;
for (client_id, client) in session.clients.iter_mut() {
if !client.paced_snapshot_due
|| now.duration_since(client.last_frame_sent) < frame_interval
{
continue;
}
client.send_seq += 1;
client.epoch_packets = client.epoch_packets.saturating_add(1);
let frame = Frame {
session: session_name.clone(),
output_seq,
bytes: snapshot.clone(),
snapshot: true,
closed: false,
};
let body = protocol::to_body(&frame)?;
let packet = protocol::encode_encrypted(
PacketKind::Frame,
*client_id,
client.send_seq,
client.last_acked,
&client.session_key,
SERVER_TO_CLIENT,
&body,
)?;
while client.pending.len() >= retransmit_window {
client.pending.pop_front();
}
client.pending.push_back(PendingFrame {
output_seq,
packet: packet.clone(),
last_sent: now,
attempts: 0,
});
client.last_frame_sent = now;
client.paced_snapshot_due = false;
sends.push((client.endpoint, packet));
}
}
sends
};
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
}
Ok(())
}
/// How long the previous epoch's key is retained after a rekey so in-flight
/// pre-rekey packets still decrypt instead of being dropped as fatal.
const REKEY_PREVIOUS_KEY_GRACE_SECS: u64 = 5;
@@ -4077,8 +3973,6 @@ mod tests {
rows: 24,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -4395,8 +4289,6 @@ mod tests {
rows: 24,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -4491,8 +4383,6 @@ mod tests {
rows: 24,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::from([42]),
@@ -4562,7 +4452,7 @@ mod tests {
}
#[tokio::test]
async fn output_pacing_coalesces_fast_frames_into_snapshot() {
async fn live_terminal_output_is_never_replaced_by_paced_snapshots() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
let config = ServerConfig {
output_frame_interval_ms: 1000,
@@ -4614,25 +4504,6 @@ mod tests {
let frame: Frame = protocol::from_body(&plain).unwrap();
assert!(!frame.snapshot);
assert_eq!(frame.bytes, b"first");
assert!(
tokio::time::timeout(Duration::from_millis(30), receiver.recv_from(&mut buf))
.await
.is_err(),
"second frame should be coalesced"
);
{
let mut locked = state.lock().unwrap();
let client = locked
.sessions
.get_mut("test")
.unwrap()
.clients
.get_mut(&client_id)
.unwrap();
client.last_frame_sent = Instant::now() - Duration::from_secs(2);
}
flush_paced_snapshots(&state, &sender).await.unwrap();
let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
.await
.unwrap()
@@ -4640,8 +4511,9 @@ mod tests {
let packet = protocol::decode(&buf[..n]).unwrap();
let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
let frame: Frame = protocol::from_body(&plain).unwrap();
assert!(frame.snapshot);
assert!(!frame.snapshot);
assert_eq!(frame.output_seq, 2);
assert_eq!(frame.bytes, b"second");
}
#[test]