Compare commits

...
11 Commits
Author SHA1 Message Date
DuProcess f8693f08b5 Release v1.0.0-rc49
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
2026-07-17 21:39:01 -04:00
DuProcess 26532fc0e1 Stop idle keepalives from repainting terminals
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
2026-07-17 21:36:44 -04:00
DuProcess c5f699a6ef Release v1.0.0-rc48
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
2026-07-17 21:25:28 -04:00
DuProcess 60403ba4c3 Stop repainting healthy idle TUIs
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
2026-07-17 21:19:31 -04:00
DuProcess 833ac1082f Release v1.0.0-rc47
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
2026-07-17 21:11:29 -04:00
DuProcess 97cf165527 Recover from sustained terminal backpressure
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
2026-07-17 21:06:26 -04:00
DuProcess 8f2d57d95e Exercise terminal reconnects across platforms
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
2026-07-17 21:00:20 -04:00
DuProcess 24180c5092 Release v1.0.0-rc46
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
2026-07-17 20:54:02 -04:00
DuProcess 0fdfc0ee22 Keep terminal backpressure test MTU safe
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
2026-07-17 20:48:50 -04:00
DuProcess 5dceb2792d Decouple terminal rendering from transport
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
2026-07-17 20:47:42 -04:00
DuProcess 58ac974fe2 Harden Windows terminal byte handling
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
2026-07-17 20:33:44 -04:00
4 changed files with 1335 additions and 126 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]] [[package]]
name = "dosh" name = "dosh"
version = "1.0.0-rc45" version = "1.0.0-rc49"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"base64", "base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "dosh" name = "dosh"
version = "1.0.0-rc45" version = "1.0.0-rc49"
edition = "2024" edition = "2024"
license = "MIT" license = "MIT"
+256 -124
View File
@@ -77,7 +77,6 @@ const POST_SUBMIT_ALL_INPUT_HOLD: Duration = Duration::from_millis(120);
const STALE_TERMINAL_INPUT_AFTER: Duration = Duration::from_secs(2); const STALE_TERMINAL_INPUT_AFTER: Duration = Duration::from_secs(2);
const POST_RECONNECT_STALE_INPUT_GRACE: Duration = Duration::from_secs(5); const POST_RECONNECT_STALE_INPUT_GRACE: Duration = Duration::from_secs(5);
const FOCUS_REPAINT_COOLDOWN: Duration = Duration::from_secs(1); const FOCUS_REPAINT_COOLDOWN: Duration = Duration::from_secs(1);
const ALT_SCREEN_IDLE_REPAINT_AFTER: Duration = Duration::from_secs(15);
const LOCAL_SLEEP_REPAINT_AFTER: Duration = Duration::from_secs(5); const LOCAL_SLEEP_REPAINT_AFTER: Duration = Duration::from_secs(5);
const LOCAL_SLEEP_REPAINT_RETRY_AFTER: Duration = Duration::from_secs(1); const LOCAL_SLEEP_REPAINT_RETRY_AFTER: Duration = Duration::from_secs(1);
const LOCAL_SLEEP_REPAINT_RETRY_WINDOW: Duration = Duration::from_secs(10); const LOCAL_SLEEP_REPAINT_RETRY_WINDOW: Duration = Duration::from_secs(10);
@@ -6560,6 +6559,7 @@ async fn run_terminal(
let mut winch = let mut winch =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::window_change()).ok(); tokio::signal::unix::signal(tokio::signal::unix::SignalKind::window_change()).ok();
let mut frame_buffer = FrameBuffer::default(); let mut frame_buffer = FrameBuffer::default();
let mut accepted_output_seq = cred.last_rendered_seq;
// Resolve the prediction display policy (off / experimental / always). An // Resolve the prediction display policy (off / experimental / always). An
// env var wins for ad-hoc tuning; otherwise the client config's // env var wins for ad-hoc tuning; otherwise the client config's
// `predict_mode` provides the persistent default. Predictions only run in a // `predict_mode` provides the persistent default. Predictions only run in a
@@ -6568,6 +6568,9 @@ async fn run_terminal(
predict && cred.mode != "view-only" && !forward_only, predict && cred.mode != "view-only" && !forward_only,
predict_mode, predict_mode,
); );
let mut frame_renderer = TerminalFrameRenderer::new()?;
let mut render_resync_needed = false;
let mut overflow_closed_frame: Option<Frame> = None;
// Non-destructive disconnect status line. Off in forward-only mode (no TTY to // Non-destructive disconnect status line. Off in forward-only mode (no TTY to
// draw on) and respecting the client config (default on, env override). // draw on) and respecting the client config (default on, env override).
let mut disconnect_status = DisconnectStatus::new(resolve_disconnect_status() && !forward_only); let mut disconnect_status = DisconnectStatus::new(resolve_disconnect_status() && !forward_only);
@@ -6610,9 +6613,8 @@ async fn run_terminal(
let mut startup_input_hold_until: Option<Instant> = None; let mut startup_input_hold_until: Option<Instant> = None;
let mut startup_gate_mode = StartupGateMode::HoldControl; let mut startup_gate_mode = StartupGateMode::HoldControl;
let mut stale_terminal_input_suppress_until: Option<Instant> = None; let mut stale_terminal_input_suppress_until: Option<Instant> = None;
let mut last_terminal_frame_at = Instant::now();
let mut last_focus_repaint_at = Instant::now() - FOCUS_REPAINT_COOLDOWN; let mut last_focus_repaint_at = Instant::now() - FOCUS_REPAINT_COOLDOWN;
let mut last_idle_repaint_attempt_at = Instant::now() - ALT_SCREEN_IDLE_REPAINT_AFTER; let mut last_idle_repaint_attempt_at = Instant::now() - LOCAL_SLEEP_REPAINT_RETRY_AFTER;
let mut last_status_tick_at = Instant::now(); let mut last_status_tick_at = Instant::now();
let mut wake_repaint_retry_until: Option<Instant> = None; let mut wake_repaint_retry_until: Option<Instant> = None;
if let Some(frame) = first_frame { if let Some(frame) = first_frame {
@@ -6620,7 +6622,6 @@ async fn run_terminal(
render_frame(&frame)?; render_frame(&frame)?;
note_snapshot_rendered(&frame, &mut disconnect_status, &mut status_restore_pending); note_snapshot_rendered(&frame, &mut disconnect_status, &mut status_restore_pending);
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
} }
if frame.closed { if frame.closed {
return Ok(()); return Ok(());
@@ -6673,10 +6674,39 @@ async fn run_terminal(
let mut recv_buf = vec![0u8; 65535]; let mut recv_buf = vec![0u8; 65535];
let mut detach_requested = false; let mut detach_requested = false;
loop { loop {
accepted_output_seq = accepted_output_seq.max(cred.last_rendered_seq);
predictor.set_output_backpressured(frame_renderer.has_pending() || render_resync_needed);
if detach_requested { if detach_requested {
break; break;
} }
tokio::select! { tokio::select! {
rendered = frame_renderer.recv(), if frame_renderer.has_pending() => {
let frame = rendered?;
predictor.set_output_backpressured(
frame_renderer.has_pending() || render_resync_needed,
);
cred.last_rendered_seq = cred.last_rendered_seq.max(frame.output_seq);
note_snapshot_rendered(
&frame,
&mut disconnect_status,
&mut status_restore_pending,
);
wake_repaint_retry_until = None;
send_ack(&socket, addr, &cred, &mut send_seq).await?;
if frame.closed {
return Ok(());
}
if !frame_renderer.has_pending()
&& let Some(frame) = overflow_closed_frame.take()
{
render_resync_needed = false;
predictor.observe_output(&frame.bytes);
anyhow::ensure!(
frame_renderer.enqueue(frame)?.is_none(),
"terminal renderer remained full after draining"
);
}
}
stdin_msg = stdin_rx.recv() => { stdin_msg = stdin_rx.recv() => {
match stdin_msg { match stdin_msg {
Some(mut bytes) => { Some(mut bytes) => {
@@ -6700,7 +6730,8 @@ async fn run_terminal(
forward_only, forward_only,
input_status_tick_gap, input_status_tick_gap,
last_packet_at.elapsed(), last_packet_at.elapsed(),
) { ) && !frame_renderer.has_pending()
{
let reconnect_started_at = Instant::now(); let reconnect_started_at = Instant::now();
if let Some(deadline) = if let Some(deadline) =
wake_repaint_retry_deadline(reconnect_started_at, input_status_tick_gap) wake_repaint_retry_deadline(reconnect_started_at, input_status_tick_gap)
@@ -6744,7 +6775,6 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
last_packet_at = Instant::now(); last_packet_at = Instant::now();
last_focus_repaint_at = Instant::now(); last_focus_repaint_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
@@ -6764,6 +6794,7 @@ async fn run_terminal(
let saw_focus_in = input_contains_focus_in(&bytes); let saw_focus_in = input_contains_focus_in(&bytes);
if !forward_only if !forward_only
&& !refreshed_before_input && !refreshed_before_input
&& !frame_renderer.has_pending()
&& saw_focus_in && saw_focus_in
&& last_focus_repaint_at.elapsed() >= FOCUS_REPAINT_COOLDOWN && last_focus_repaint_at.elapsed() >= FOCUS_REPAINT_COOLDOWN
{ {
@@ -6803,7 +6834,6 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
last_packet_at = Instant::now(); last_packet_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
flush_pending_user_input( flush_pending_user_input(
@@ -6997,7 +7027,8 @@ async fn run_terminal(
&mut pending_user_input_bytes, &mut pending_user_input_bytes,
bytes, bytes,
)?; )?;
if let Some(frame) = reconnect( if !frame_renderer.has_pending()
&& let Some(frame) = reconnect(
&socket, &socket,
&mut cred, &mut cred,
&mut send_seq, &mut send_seq,
@@ -7026,7 +7057,6 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
} }
last_packet_at = Instant::now(); last_packet_at = Instant::now();
flush_pending_user_input( flush_pending_user_input(
@@ -7075,7 +7105,9 @@ async fn run_terminal(
maybe_send_resize(&socket, addr, &cred, &mut send_seq, &mut last_size).await?; maybe_send_resize(&socket, addr, &cred, &mut send_seq, &mut last_size).await?;
} }
_ = frame_gap_tick.tick() => { _ = frame_gap_tick.tick() => {
if frame_buffer.resync_due() if !render_resync_needed
&& !frame_renderer.has_pending()
&& frame_buffer.resync_due()
&& let Some(frame) = reconnect( && let Some(frame) = reconnect(
&socket, &socket,
&mut cred, &mut cred,
@@ -7098,7 +7130,6 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
} }
last_packet_at = Instant::now(); last_packet_at = Instant::now();
@@ -7138,6 +7169,9 @@ async fn run_terminal(
} }
}); });
let Ok(plain) = decrypted else { let Ok(plain) = decrypted else {
if frame_renderer.has_pending() {
continue;
}
if let Some(frame) = reconnect( if let Some(frame) = reconnect(
&socket, &socket,
&mut cred, &mut cred,
@@ -7160,7 +7194,6 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
} }
last_packet_at = Instant::now(); last_packet_at = Instant::now();
@@ -7185,19 +7218,36 @@ async fn run_terminal(
continue; continue;
}; };
last_packet_at = Instant::now(); last_packet_at = Instant::now();
let frames = frame_buffer.accept(frame, &mut cred.last_rendered_seq); let frames = frame_buffer.accept(frame, &mut accepted_output_seq);
for frame in frames { for frame in frames {
predictor.clear_pending()?; predictor.clear_pending()?;
if !forward_only { if !forward_only {
render_frame(&frame)?;
note_snapshot_rendered(
&frame,
&mut disconnect_status,
&mut status_restore_pending,
);
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
if render_resync_needed {
if frame.closed {
overflow_closed_frame = Some(frame);
}
} else if let Some(frame) = frame_renderer.enqueue(frame)? {
dosh::trace::event(
"client.render_queue_overflow",
&[
("output_seq", frame.output_seq.to_string()),
("bytes", frame.bytes.len().to_string()),
],
);
dosh::trace::health_event(
"client.render_queue_overflow",
&[("output_seq", frame.output_seq.to_string())],
);
render_resync_needed = true;
if frame.closed {
overflow_closed_frame = Some(frame);
}
}
predictor.set_output_backpressured(
frame_renderer.has_pending() || render_resync_needed,
);
flush_startup_input_if_ready( flush_startup_input_if_ready(
&socket, &socket,
addr, addr,
@@ -7209,12 +7259,15 @@ async fn run_terminal(
) )
.await?; .await?;
} }
if frame.closed { else if frame.closed {
send_ack(&socket, addr, &cred, &mut send_seq).await?; send_ack(&socket, addr, &cred, &mut send_seq).await?;
return Ok(()); return Ok(());
} }
} }
send_ack(&socket, addr, &cred, &mut send_seq).await?; if forward_only {
cred.last_rendered_seq = accepted_output_seq;
send_ack(&socket, addr, &cred, &mut send_seq).await?;
}
} }
PacketKind::Pong => { PacketKind::Pong => {
if protocol::decrypt_body( if protocol::decrypt_body(
@@ -7302,7 +7355,7 @@ async fn run_terminal(
} }
PacketKind::AttachReject => { PacketKind::AttachReject => {
let reject: AttachReject = protocol::from_body(&packet.body)?; let reject: AttachReject = protocol::from_body(&packet.body)?;
if reject.reason == "unknown client" { if reject.reason == "unknown client" && !frame_renderer.has_pending() {
if let Some(frame) = reconnect( if let Some(frame) = reconnect(
&socket, &socket,
&mut cred, &mut cred,
@@ -7325,7 +7378,6 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
} }
last_packet_at = Instant::now(); last_packet_at = Instant::now();
@@ -7918,8 +7970,10 @@ async fn run_terminal(
} }
let mut repainted_this_tick = false; let mut repainted_this_tick = false;
let stale = last_packet_at.elapsed(); let stale = last_packet_at.elapsed();
if status_restore_pending if !frame_renderer.has_pending()
|| stale >= Duration::from_secs(reconnect_timeout_secs.max(1)) && (render_resync_needed
|| status_restore_pending
|| stale >= Duration::from_secs(reconnect_timeout_secs.max(1)))
{ {
if let Some(frame) = reconnect( if let Some(frame) = reconnect(
&socket, &socket,
@@ -7943,10 +7997,10 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
last_idle_repaint_attempt_at = Instant::now(); last_idle_repaint_attempt_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
repainted_this_tick = true; repainted_this_tick = true;
render_resync_needed = false;
} }
last_packet_at = Instant::now(); last_packet_at = Instant::now();
flush_pending_user_input( flush_pending_user_input(
@@ -7977,7 +8031,9 @@ async fn run_terminal(
// a latency spike (or recovery) flips speculation on/off promptly // a latency spike (or recovery) flips speculation on/off promptly
// without waiting for the next keystroke to drive `redraw`. // without waiting for the next keystroke to drive `redraw`.
if !forward_only { if !forward_only {
predictor.refresh_policy()?; if !frame_renderer.has_pending() && !render_resync_needed {
predictor.refresh_policy()?;
}
flush_startup_input_if_ready( flush_startup_input_if_ready(
&socket, &socket,
addr, addr,
@@ -7989,9 +8045,7 @@ async fn run_terminal(
) )
.await?; .await?;
let now = Instant::now(); let now = Instant::now();
if !repainted_this_tick && should_repaint_idle_terminal( if !repainted_this_tick && !render_resync_needed && !frame_renderer.has_pending() && should_repaint_idle_terminal(
predictor.alternate_screen,
last_terminal_frame_at,
last_idle_repaint_attempt_at, last_idle_repaint_attempt_at,
status_tick_gap, status_tick_gap,
wake_repaint_retry_until, wake_repaint_retry_until,
@@ -8027,7 +8081,6 @@ async fn run_terminal(
&mut status_restore_pending, &mut status_restore_pending,
); );
predictor.observe_output(&frame.bytes); predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
last_packet_at = Instant::now(); last_packet_at = Instant::now();
wake_repaint_retry_until = None; wake_repaint_retry_until = None;
flush_pending_user_input( flush_pending_user_input(
@@ -8047,7 +8100,10 @@ async fn run_terminal(
// on how long the link has been silent (recomputed after any // on how long the link has been silent (recomputed after any
// reconnect attempt above may have reset `last_packet_at`). // reconnect attempt above may have reset `last_packet_at`).
if !forward_only { if !forward_only {
let action = if predictor.alternate_screen { let action = if predictor.alternate_screen
|| frame_renderer.has_pending()
|| render_resync_needed
{
disconnect_status.on_suppressed() disconnect_status.on_suppressed()
} else { } else {
disconnect_status.on_tick(last_packet_at.elapsed()) disconnect_status.on_tick(last_packet_at.elapsed())
@@ -8599,8 +8655,6 @@ fn strip_terminal_focus_reports(bytes: &[u8]) -> Vec<u8> {
} }
fn should_repaint_idle_terminal( fn should_repaint_idle_terminal(
alternate_screen: bool,
last_terminal_frame_at: Instant,
last_attempt_at: Instant, last_attempt_at: Instant,
status_tick_gap: Duration, status_tick_gap: Duration,
wake_repaint_retry_until: Option<Instant>, wake_repaint_retry_until: Option<Instant>,
@@ -8608,12 +8662,8 @@ fn should_repaint_idle_terminal(
) -> bool { ) -> bool {
let sleep_wake_gap = status_tick_gap >= LOCAL_SLEEP_REPAINT_AFTER; let sleep_wake_gap = status_tick_gap >= LOCAL_SLEEP_REPAINT_AFTER;
let wake_retry_active = wake_repaint_retry_until.is_some_and(|deadline| now < deadline); let wake_retry_active = wake_repaint_retry_until.is_some_and(|deadline| now < deadline);
let stale_alternate_screen = alternate_screen (sleep_wake_gap || wake_retry_active)
&& now.duration_since(last_terminal_frame_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER; && now.duration_since(last_attempt_at) >= LOCAL_SLEEP_REPAINT_RETRY_AFTER
if sleep_wake_gap || wake_retry_active {
return now.duration_since(last_attempt_at) >= LOCAL_SLEEP_REPAINT_RETRY_AFTER;
}
stale_alternate_screen && now.duration_since(last_attempt_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
} }
fn arm_stale_terminal_input_suppression(suppress_until: &mut Option<Instant>) { fn arm_stale_terminal_input_suppression(suppress_until: &mut Option<Instant>) {
@@ -9747,6 +9797,9 @@ struct PredictedCell {
struct Predictor { struct Predictor {
mode: PredictMode, mode: PredictMode,
enabled: bool, enabled: bool,
/// Suppress local display writes while authoritative output is waiting on a
/// slow terminal renderer. Input prediction state is still maintained.
output_backpressured: bool,
/// True while the server is in the alternate screen (a full-screen TUI such /// True while the server is in the alternate screen (a full-screen TUI such
/// as vim/htop); we never speculate there because we cannot model arbitrary /// as vim/htop); we never speculate there because we cannot model arbitrary
/// cursor addressing safely. /// cursor addressing safely.
@@ -9803,6 +9856,7 @@ impl Predictor {
Self { Self {
mode, mode,
enabled: enabled && mode != PredictMode::Off, enabled: enabled && mode != PredictMode::Off,
output_backpressured: false,
alternate_screen: false, alternate_screen: false,
mouse_tracking: TerminalMouseMode::None, mouse_tracking: TerminalMouseMode::None,
output_parse_tail: Vec::new(), output_parse_tail: Vec::new(),
@@ -10067,6 +10121,9 @@ impl Predictor {
/// Whether we should *display* predictions right now under the active policy. /// Whether we should *display* predictions right now under the active policy.
fn should_display(&self) -> bool { fn should_display(&self) -> bool {
if self.output_backpressured {
return false;
}
match self.mode { match self.mode {
PredictMode::Off => false, PredictMode::Off => false,
PredictMode::Always => true, PredictMode::Always => true,
@@ -10085,6 +10142,10 @@ impl Predictor {
} }
} }
fn set_output_backpressured(&mut self, value: bool) {
self.output_backpressured = value;
}
/// Update the SRTT/flag hysteresis latches from the current estimate. /// Update the SRTT/flag hysteresis latches from the current estimate.
fn update_triggers(&mut self) { fn update_triggers(&mut self) {
let srtt = self.srtt_ms.unwrap_or(0.0); let srtt = self.srtt_ms.unwrap_or(0.0);
@@ -10500,6 +10561,83 @@ async fn detach_once(socket: &UdpSocket, cred: &CachedCredential, seq: u64) -> R
Ok(()) Ok(())
} }
const TERMINAL_RENDER_QUEUE_CAPACITY: usize = 256;
/// Keeps potentially slow console writes off the UDP event loop. Frame ACKs are
/// emitted only after `recv` reports completion, so the server never retires
/// output that has merely been queued locally rather than displayed.
struct TerminalFrameRenderer {
jobs: Option<mpsc::Sender<Frame>>,
completed: mpsc::UnboundedReceiver<Result<Frame, String>>,
pending: usize,
thread: Option<std::thread::JoinHandle<()>>,
}
impl TerminalFrameRenderer {
fn new() -> Result<Self> {
let (job_tx, mut job_rx) = mpsc::channel::<Frame>(TERMINAL_RENDER_QUEUE_CAPACITY);
let (completed_tx, completed_rx) = mpsc::unbounded_channel();
let thread = std::thread::Builder::new()
.name("dosh-terminal-render".to_string())
.spawn(move || {
while let Some(frame) = job_rx.blocking_recv() {
let result = render_frame(&frame)
.map(|()| frame)
.map_err(|err| format!("render terminal frame: {err:#}"));
if completed_tx.send(result).is_err() {
break;
}
}
})?;
Ok(Self {
jobs: Some(job_tx),
completed: completed_rx,
pending: 0,
thread: Some(thread),
})
}
/// Returns the frame unchanged when the bounded queue is full. The caller
/// can continue servicing transport traffic and request a snapshot once the
/// renderer drains instead of turning local display backpressure into a
/// disconnected session.
fn enqueue(&mut self, frame: Frame) -> Result<Option<Frame>> {
let Some(jobs) = self.jobs.as_ref() else {
return Err(anyhow!("terminal renderer is closed"));
};
match jobs.try_send(frame) {
Ok(()) => {
self.pending += 1;
Ok(None)
}
Err(mpsc::error::TrySendError::Full(frame)) => Ok(Some(frame)),
Err(mpsc::error::TrySendError::Closed(_)) => Err(anyhow!("terminal renderer stopped")),
}
}
fn has_pending(&self) -> bool {
self.pending > 0
}
async fn recv(&mut self) -> Result<Frame> {
let result =
self.completed.recv().await.ok_or_else(|| {
anyhow!("terminal renderer stopped before completing queued output")
})?;
self.pending = self.pending.saturating_sub(1);
result.map_err(anyhow::Error::msg)
}
}
impl Drop for TerminalFrameRenderer {
fn drop(&mut self) {
self.jobs.take();
if let Some(thread) = self.thread.take() {
let _ = thread.join();
}
}
}
fn render_frame(frame: &Frame) -> Result<()> { fn render_frame(frame: &Frame) -> Result<()> {
let mut stdout = std::io::stdout(); let mut stdout = std::io::stdout();
stdout.write_all(&render_frame_bytes(frame))?; stdout.write_all(&render_frame_bytes(frame))?;
@@ -10538,9 +10676,12 @@ const TERMINAL_SNAPSHOT_RESET: &[u8] = concat!(
.as_bytes(); .as_bytes();
/// Seconds of silence from the server before the disconnect status line appears. /// Seconds of silence from the server before the disconnect status line appears.
/// Short enough to give quick feedback on a lost link, long enough that a normal /// Short enough to give quick feedback on a lost link, but later than the first
/// idle period (the run loop only pings after 2s of quiet) never flashes it. /// idle ping. Using the same threshold as the first ping races its Pong: the bar
const DISCONNECT_STATUS_THRESHOLD_SECS: u64 = 2; /// is painted just before the authenticated reply arrives, which then requests a
/// snapshot to restore row 1 and turns every healthy idle connection into a
/// periodic reconnect loop.
const DISCONNECT_STATUS_THRESHOLD_SECS: u64 = 3;
/// Mosh-style disconnect status bar with snapshot-backed restoration. /// Mosh-style disconnect status bar with snapshot-backed restoration.
/// ///
@@ -10845,6 +10986,8 @@ fn windows_vt_input_mode(mode: u32) -> u32 {
struct WindowsConsoleModeGuard { struct WindowsConsoleModeGuard {
input: Option<(windows_sys::Win32::Foundation::HANDLE, u32)>, input: Option<(windows_sys::Win32::Foundation::HANDLE, u32)>,
output: Option<(windows_sys::Win32::Foundation::HANDLE, u32)>, output: Option<(windows_sys::Win32::Foundation::HANDLE, u32)>,
input_code_page: Option<u32>,
output_code_page: Option<u32>,
} }
#[cfg(windows)] #[cfg(windows)]
@@ -10853,51 +10996,91 @@ impl WindowsConsoleModeGuard {
unsafe { unsafe {
use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE; use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
use windows_sys::Win32::System::Console::{ use windows_sys::Win32::System::Console::{
GetConsoleMode, GetStdHandle, STD_INPUT_HANDLE, STD_OUTPUT_HANDLE, SetConsoleMode, GetConsoleCP, GetConsoleMode, GetConsoleOutputCP, GetStdHandle, STD_INPUT_HANDLE,
STD_OUTPUT_HANDLE, SetConsoleCP, SetConsoleMode, SetConsoleOutputCP,
};
const CP_UTF8: u32 = 65001;
let mut guard = Self {
input: None,
output: None,
input_code_page: None,
output_code_page: None,
}; };
let input_handle = GetStdHandle(STD_INPUT_HANDLE); let input_handle = GetStdHandle(STD_INPUT_HANDLE);
let input = if input_handle.is_null() || input_handle == INVALID_HANDLE_VALUE { if !input_handle.is_null() && input_handle != INVALID_HANDLE_VALUE {
None
} else {
let mut original = 0u32; let mut original = 0u32;
if GetConsoleMode(input_handle, &mut original) == 0 { if GetConsoleMode(input_handle, &mut original) != 0 {
None
} else {
let desired = windows_vt_input_mode(original); let desired = windows_vt_input_mode(original);
if desired != original && SetConsoleMode(input_handle, desired) == 0 { if desired != original && SetConsoleMode(input_handle, desired) == 0 {
return Err(std::io::Error::last_os_error()) return Err(std::io::Error::last_os_error())
.context("enable Windows virtual-terminal input mode"); .context("enable Windows virtual-terminal input mode");
} }
Some((input_handle, original)) guard.input = Some((input_handle, original));
let original_code_page = GetConsoleCP();
if original_code_page == 0 {
let err = std::io::Error::last_os_error();
guard.restore();
return Err(err).context("read Windows console input code page");
}
guard.input_code_page = Some(original_code_page);
if original_code_page != CP_UTF8 && SetConsoleCP(CP_UTF8) == 0 {
let err = std::io::Error::last_os_error();
guard.restore();
return Err(err).context("set Windows console input to UTF-8");
}
} }
}; }
let output_handle = GetStdHandle(STD_OUTPUT_HANDLE); let output_handle = GetStdHandle(STD_OUTPUT_HANDLE);
let output = if output_handle.is_null() || output_handle == INVALID_HANDLE_VALUE { if !output_handle.is_null() && output_handle != INVALID_HANDLE_VALUE {
None
} else {
let mut original = 0u32; let mut original = 0u32;
if GetConsoleMode(output_handle, &mut original) == 0 { if GetConsoleMode(output_handle, &mut original) != 0 {
None
} else {
let desired = windows_vt_output_mode(original); let desired = windows_vt_output_mode(original);
if desired != original && SetConsoleMode(output_handle, desired) == 0 { if desired != original && SetConsoleMode(output_handle, desired) == 0 {
let err = std::io::Error::last_os_error(); let err = std::io::Error::last_os_error();
if let Some((handle, mode)) = input { guard.restore();
let _ = SetConsoleMode(handle, mode);
}
return Err(err).context("enable Windows virtual-terminal output mode"); return Err(err).context("enable Windows virtual-terminal output mode");
} }
Some((output_handle, original)) guard.output = Some((output_handle, original));
}
};
Ok(Self { input, output }) let original_code_page = GetConsoleOutputCP();
if original_code_page == 0 {
let err = std::io::Error::last_os_error();
guard.restore();
return Err(err).context("read Windows console output code page");
}
guard.output_code_page = Some(original_code_page);
if original_code_page != CP_UTF8 && SetConsoleOutputCP(CP_UTF8) == 0 {
let err = std::io::Error::last_os_error();
guard.restore();
return Err(err).context("set Windows console output to UTF-8");
}
}
}
Ok(guard)
} }
} }
fn restore(&self) { fn restore(&self) {
if let Some(code_page) = self.output_code_page {
unsafe {
use windows_sys::Win32::System::Console::SetConsoleOutputCP;
let _ = SetConsoleOutputCP(code_page);
}
}
if let Some(code_page) = self.input_code_page {
unsafe {
use windows_sys::Win32::System::Console::SetConsoleCP;
let _ = SetConsoleCP(code_page);
}
}
if let Some((handle, mode)) = self.input { if let Some((handle, mode)) = self.input {
unsafe { unsafe {
use windows_sys::Win32::System::Console::SetConsoleMode; use windows_sys::Win32::System::Console::SetConsoleMode;
@@ -11005,8 +11188,8 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{ use super::{
ALT_SCREEN_IDLE_REPAINT_AFTER, CachedCredential, DisconnectStatus, DynamicForward, CachedCredential, DisconnectStatus, DynamicForward, FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer,
FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, LOCAL_SLEEP_REPAINT_AFTER, LOCAL_SLEEP_REPAINT_AFTER, LOCAL_SLEEP_REPAINT_RETRY_AFTER,
LOCAL_SLEEP_REPAINT_RETRY_WINDOW, LocalForward, MAX_PENDING_USER_INPUT_BYTES, LOCAL_SLEEP_REPAINT_RETRY_WINDOW, LocalForward, MAX_PENDING_USER_INPUT_BYTES,
NativeIdentityContext, POST_RECONNECT_STALE_INPUT_GRACE, POST_SUBMIT_ALL_INPUT_HOLD, NativeIdentityContext, POST_RECONNECT_STALE_INPUT_GRACE, POST_SUBMIT_ALL_INPUT_HOLD,
PendingStreamControl, PendingStreamOpen, PendingWindowAdjust, PredictMode, Predictor, PendingStreamControl, PendingStreamOpen, PendingWindowAdjust, PredictMode, Predictor,
@@ -14277,86 +14460,35 @@ mod tests {
} }
#[test] #[test]
fn idle_repaint_runs_for_stale_alternate_screen_or_sleep_gap() { fn idle_repaint_runs_only_after_sleep_or_an_armed_wake_retry() {
let now = Instant::now(); let now = Instant::now();
let stale = now - ALT_SCREEN_IDLE_REPAINT_AFTER - Duration::from_secs(1); let stale = now - LOCAL_SLEEP_REPAINT_RETRY_AFTER - Duration::from_secs(1);
let recent = now - Duration::from_secs(1);
let just_attempted = now - Duration::from_millis(500); let just_attempted = now - Duration::from_millis(500);
assert!(should_repaint_idle_terminal( assert!(!should_repaint_idle_terminal(
true,
stale,
stale, stale,
Duration::from_secs(1), Duration::from_secs(1),
None, None,
now now
)); ));
assert!(should_repaint_idle_terminal( assert!(should_repaint_idle_terminal(
false,
recent,
stale, stale,
LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1), LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
None, None,
now now
)); ));
assert!(should_repaint_idle_terminal( assert!(should_repaint_idle_terminal(
false, stale,
recent, Duration::from_secs(1),
recent,
LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW), Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
now now
)); ));
assert!(!should_repaint_idle_terminal( assert!(!should_repaint_idle_terminal(
false,
recent,
just_attempted, just_attempted,
LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1), LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW), Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
now now
)); ));
assert!(!should_repaint_idle_terminal( assert!(!should_repaint_idle_terminal(
false,
stale,
stale,
Duration::from_secs(1),
None,
now
));
assert!(!should_repaint_idle_terminal(
true,
recent,
stale,
Duration::from_secs(1),
None,
now
));
assert!(should_repaint_idle_terminal(
true,
stale,
recent,
LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
None,
now
));
assert!(should_repaint_idle_terminal(
false,
recent,
stale,
Duration::from_secs(1),
Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
now
));
assert!(!should_repaint_idle_terminal(
false,
recent,
just_attempted,
Duration::from_secs(1),
Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
now
));
assert!(!should_repaint_idle_terminal(
false,
recent,
stale, stale,
Duration::from_secs(1), Duration::from_secs(1),
Some(now - Duration::from_millis(1)), Some(now - Duration::from_millis(1)),
File diff suppressed because it is too large Load Diff