use blit_alacritty::{SearchResult as AlacrittySearchResult, TerminalDriver as AlacrittyDriver};
use blit_compositor::{CompositorCommand, CompositorEvent, CompositorHandle};
use blit_remote::{
C2S_ACK, C2S_AUDIO_SUBSCRIBE, C2S_AUDIO_UNSUBSCRIBE, C2S_CLIENT_FEATURES, C2S_CLIENT_METRICS,
C2S_CLIPBOARD_GET, C2S_CLIPBOARD_LIST, C2S_CLIPBOARD_SET, C2S_CLOSE, C2S_COPY_RANGE,
C2S_CREATE, C2S_CREATE_AT, C2S_CREATE_N, C2S_CREATE2, C2S_DISPLAY_RATE, C2S_FOCUS, C2S_INPUT,
C2S_KILL, C2S_MOUSE, C2S_PING, C2S_QUIT, C2S_READ, C2S_RESIZE, C2S_RESTART, C2S_SCROLL,
C2S_SEARCH, C2S_SUBSCRIBE, C2S_SURFACE_ACK, C2S_SURFACE_CAPTURE, C2S_SURFACE_CLOSE,
C2S_SURFACE_FOCUS, C2S_SURFACE_INPUT, C2S_SURFACE_LIST, C2S_SURFACE_POINTER,
C2S_SURFACE_POINTER_AXIS, C2S_SURFACE_RESIZE, C2S_SURFACE_SUBSCRIBE, C2S_SURFACE_TEXT,
C2S_SURFACE_UNSUBSCRIBE, C2S_UNSUBSCRIBE, CAPTURE_FORMAT_AVIF, CAPTURE_FORMAT_PNG,
CREATE2_HAS_COMMAND, CREATE2_HAS_SRC_PTY, FEATURE_AUDIO, FEATURE_COMPOSITOR,
FEATURE_COPY_RANGE, FEATURE_CREATE_NONCE, FEATURE_RESIZE_BATCH, FEATURE_RESTART, FrameState,
READ_ANSI, READ_TAIL, S2C_CLOSED, S2C_CREATED, S2C_CREATED_N, S2C_LIST, S2C_PING, S2C_QUIT,
S2C_READY, S2C_SEARCH_RESULTS, S2C_SURFACE_CAPTURE, S2C_SURFACE_LIST, S2C_TEXT, S2C_TITLE,
SURFACE_FRAME_FLAG_KEYFRAME, build_update_msg, msg_hello, msg_s2c_clipboard_content,
msg_s2c_clipboard_list, msg_surface_app_id, msg_surface_created, msg_surface_destroyed,
msg_surface_encoder, msg_surface_frame, msg_surface_resized, msg_surface_title,
};
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::{Mutex, Notify, mpsc};
#[cfg(target_os = "linux")]
mod audio;
#[cfg(target_os = "linux")]
mod audio_pw;
mod gpu_libs;
mod ipc;
mod nvenc_encode;
mod pty;
mod surface_encoder;
#[cfg(target_os = "linux")]
mod vaapi_encode;
pub use ipc::{IpcListener, default_ipc_path};
use pty::{PtyHandle, PtyWriteTarget};
pub use surface_encoder::ChromaSubsampling;
use surface_encoder::SurfaceEncoder;
pub use surface_encoder::SurfaceEncoderPreference;
pub use surface_encoder::SurfaceH264EncoderPreference;
pub use surface_encoder::SurfaceQuality;
type PtyFds = Arc<std::sync::RwLock<HashMap<u16, PtyWriteTarget>>>;
pub struct Config {
pub shell: String,
pub shell_flags: String,
pub scrollback: usize,
pub ipc_path: String,
pub surface_encoders: Vec<SurfaceEncoderPreference>,
pub surface_quality: SurfaceQuality,
pub chroma: ChromaSubsampling,
pub vaapi_device: String,
#[cfg(unix)]
pub fd_channel: Option<std::os::unix::io::RawFd>,
pub verbose: bool,
/// Maximum number of concurrent client connections (0 = unlimited).
pub max_connections: usize,
/// Maximum number of PTYs across all clients (0 = unlimited).
pub max_ptys: usize,
/// Application-level ping interval. The server sends S2C_PING to every
/// client at this cadence so that transports without native keepalive
/// (WebRTC data channels) can detect dead connections. 0 = disabled.
pub ping_interval: Duration,
/// Skip compositor initialization (e.g. for share-only mode).
pub skip_compositor: bool,
}
trait PtyDriver: Send {
fn size(&self) -> (u16, u16);
fn resize(&mut self, rows: u16, cols: u16);
fn process(&mut self, data: &[u8]);
fn title(&self) -> &str;
fn search_result(&self, query: &str) -> Option<PtySearchResult>;
fn take_title_dirty(&mut self) -> bool;
fn cursor_position(&self) -> (u16, u16);
fn synced_output(&self) -> bool;
fn snapshot(&mut self, echo: bool, icanon: bool) -> FrameState;
fn scrollback_frame(&mut self, offset: usize) -> FrameState;
fn reset_modes(&mut self);
fn mouse_event(
&self,
type_: u8,
button: u8,
col: u16,
row: u16,
echo: bool,
icanon: bool,
) -> Option<Vec<u8>>;
fn get_text_range(
&self,
start_tail: u32,
start_col: u16,
end_tail: u32,
end_col: u16,
) -> String;
fn total_lines(&self) -> u32;
}
struct PtySearchResult {
score: u32,
primary_source: u8,
matched_sources: u8,
context: String,
scroll_offset: Option<usize>,
}
impl PtyDriver for AlacrittyDriver {
fn size(&self) -> (u16, u16) {
AlacrittyDriver::size(self)
}
fn resize(&mut self, rows: u16, cols: u16) {
AlacrittyDriver::resize(self, rows, cols);
}
fn process(&mut self, data: &[u8]) {
AlacrittyDriver::process(self, data);
}
fn title(&self) -> &str {
AlacrittyDriver::title(self)
}
fn search_result(&self, query: &str) -> Option<PtySearchResult> {
AlacrittyDriver::search_result(self, query).map(|result: AlacrittySearchResult| {
PtySearchResult {
score: result.score,
primary_source: result.primary_source as u8,
matched_sources: result.matched_sources,
context: result.context,
scroll_offset: result.scroll_offset,
}
})
}
fn take_title_dirty(&mut self) -> bool {
AlacrittyDriver::take_title_dirty(self)
}
fn cursor_position(&self) -> (u16, u16) {
AlacrittyDriver::cursor_position(self)
}
fn synced_output(&self) -> bool {
AlacrittyDriver::synced_output(self)
}
fn snapshot(&mut self, echo: bool, icanon: bool) -> FrameState {
AlacrittyDriver::snapshot(self, echo, icanon)
}
fn scrollback_frame(&mut self, offset: usize) -> FrameState {
AlacrittyDriver::scrollback_frame(self, offset)
}
fn reset_modes(&mut self) {
AlacrittyDriver::reset_modes(self);
}
fn mouse_event(
&self,
type_: u8,
button: u8,
col: u16,
row: u16,
echo: bool,
icanon: bool,
) -> Option<Vec<u8>> {
AlacrittyDriver::mouse_event(self, type_, button, col, row, echo, icanon)
}
fn get_text_range(
&self,
start_tail: u32,
start_col: u16,
end_tail: u32,
end_col: u16,
) -> String {
AlacrittyDriver::get_text_range(self, start_tail, start_col, end_tail, end_col)
}
fn total_lines(&self) -> u32 {
AlacrittyDriver::total_lines(self)
}
}
// Soft backpressure thresholds. The outbox channel is unbounded so messages
// are never dropped, but production is throttled (via `window_open` /
// `surface_window_open`) once either counter exceeds these limits.
const OUTBOX_SOFT_QUEUE_LIMIT_FRAMES: usize = 4;
// Must comfortably hold one keyframe at 1920x1080 from a software encoder
// (200-400 KB is typical). Setting this too low deadlocks the outbox gate
// when a single frame exceeds the cap — surface_window_open returns false
// even at outbox=1, and no new frames can ever be produced.
const OUTBOX_SOFT_QUEUE_LIMIT_BYTES: usize = 1024 * 1024;
const PREVIEW_FRAME_RESERVE: usize = 1;
const READY_FRAME_QUEUE_CAP: usize = 4;
const PTY_CHANNEL_CAPACITY: usize = 64;
const SYNC_OUTPUT_END: &[u8] = b"\x1b[?2026l";
/// Number of surface frames to send at wire speed after a keyframe request
/// (subscribe, resubscribe, or error recovery). During this burst window
/// only outbox backpressure gates delivery — the time-based pacing interval
/// is skipped. This lets bandwidth estimates ramp up quickly on high-latency
/// links instead of starving the pipeline with conservative initial rates.
const SURFACE_BURST_FRAMES: u8 = 4;
/// A chunk of data from the PTY reader, sent through a lock-free channel
/// so the reader never contends with the delivery tick for the Session mutex.
enum PtyInput {
/// Raw bytes from the PTY, with the reader's sync-scan tail for boundary
/// detection. The tick task calls `process()` + `respond_to_queries()`.
Data(Vec<u8>),
/// Data up to and including a sync-output-close (`\x1b[?2026l`).
/// Process `before` and then take a snapshot. Any bytes following the
/// boundary are sent in a subsequent `Data` or `SyncBoundary` event —
/// the reader's loop re-scans them, so this event must not try to
/// process them itself.
SyncBoundary { before: Vec<u8> },
/// The PTY fd hit EOF or an error — the child likely exited.
Eof,
}
const MAX_FRAME_SIZE: usize = 16 * 1024 * 1024;
async fn read_frame(reader: &mut (impl AsyncRead + Unpin)) -> Option<Vec<u8>> {
let mut len_buf = [0u8; 4];
reader.read_exact(&mut len_buf).await.ok()?;
let len = u32::from_le_bytes(len_buf) as usize;
if len == 0 {
return Some(vec![]);
}
if len > MAX_FRAME_SIZE {
return None;
}
let mut buf = vec![0u8; len];
reader.read_exact(&mut buf).await.ok()?;
Some(buf)
}
async fn write_frame(writer: &mut (impl AsyncWrite + Unpin), payload: &[u8]) -> bool {
if payload.len() > u32::MAX as usize {
return false;
}
let len = payload.len() as u32;
let mut buf = Vec::with_capacity(4 + payload.len());
buf.extend_from_slice(&len.to_le_bytes());
buf.extend_from_slice(payload);
writer.write_all(&buf).await.is_ok()
}
/// Largest bulk-frame payload we'll write in a single length-prefixed
/// message. Payloads above this are split into `S2C_FRAGMENT` messages
/// so audio frames can be drained between chunks, bounding the time
/// audio sits blocked behind a bulk write to (roughly) `CHUNK_BYTES /
/// network_bandwidth`. Too small and per-message overhead dominates
/// (each fragment adds an 8-byte length prefix + 2-byte fragment header);
/// too large and audio suffers head-of-line blocking again. 4 KiB keeps
/// per-chunk write time under ~4 ms even on a 1 MB/s link — well below
/// the 20 ms audio frame cadence with headroom for a handful of chunks
/// in flight.
const BULK_CHUNK_BYTES: usize = 4 * 1024;
/// Write a bulk message, draining pending audio frames between chunks.
///
/// Payloads that fit within `BULK_CHUNK_BYTES` are written as a single
/// length-prefixed frame after a pre-drain of pending audio (same as
/// before). Larger payloads are split into `S2C_FRAGMENT` messages so
/// audio frames written between fragments remain valid, complete,
/// length-prefixed messages on the wire — never interleaved inside a
/// single `read_exact`-delimited payload, which would desynchronise
/// the reader's framing.
async fn write_frame_interleaved(
writer: &mut (impl AsyncWrite + Unpin),
payload: &[u8],
audio_rx: &mut mpsc::UnboundedReceiver<Vec<u8>>,
) -> bool {
// Small message: drain any queued audio, then write as-is.
if payload.len() <= BULK_CHUNK_BYTES {
while let Ok(audio_msg) = audio_rx.try_recv() {
if !write_frame(writer, &audio_msg).await {
return false;
}
}
return write_frame(writer, payload).await;
}
// Large message: split into S2C_FRAGMENT messages, draining audio
// between each chunk. The chunks carry the original payload bytes
// verbatim (including its type byte in the first chunk); the
// receiver concatenates them and dispatches the reassembled buffer.
let mut offset = 0;
while offset < payload.len() {
while let Ok(audio_msg) = audio_rx.try_recv() {
if !write_frame(writer, &audio_msg).await {
return false;
}
}
let end = (offset + BULK_CHUNK_BYTES).min(payload.len());
let is_last = end == payload.len();
let mut frag = Vec::with_capacity(2 + (end - offset));
frag.push(blit_remote::S2C_FRAGMENT);
frag.push(if is_last {
blit_remote::FRAGMENT_FLAG_LAST
} else {
0
});
frag.extend_from_slice(&payload[offset..end]);
if !write_frame(writer, &frag).await {
return false;
}
offset = end;
}
true
}
struct Pty {
handle: PtyHandle,
driver: Box<dyn PtyDriver>,
/// Client-chosen tag set at creation time.
tag: String,
dirty: bool,
ready_frames: VecDeque<FrameState>,
/// Receives raw byte chunks from the PTY reader task without mutex contention.
byte_rx: mpsc::Receiver<PtyInput>,
reader_handle: std::thread::JoinHandle<()>,
/// Cached (echo, icanon) from tcgetattr; refreshed every ~250ms.
lflag_cache: (bool, bool),
lflag_last: Instant,
/// When we last broadcast a title update for this PTY.
last_title_send: Instant,
/// Title changed but not yet sent (debounced).
title_pending: bool,
/// The subprocess has exited but the terminal state is retained for reading.
exited: bool,
/// Exit status: WEXITSTATUS if normal exit, negative signal number if signalled,
/// EXIT_STATUS_UNKNOWN if not yet collected.
exit_status: i32,
/// Command used to create this PTY (None = default shell).
command: Option<String>,
}
impl Pty {
fn mark_dirty(&mut self) {
self.dirty = true;
}
fn clear_dirty(&mut self) {
self.dirty = false;
}
}
struct CachedSurfaceInfo {
surface_id: u16,
parent_id: u16,
width: u16,
height: u16,
title: String,
app_id: String,
}
/// Last committed pixel buffer for a surface, kept so we can re-encode a
/// keyframe for late-joining clients without going back to the compositor.
struct LastPixels {
width: u32,
height: u32,
pixels: blit_compositor::PixelData,
/// Monotonically increasing counter bumped on every SurfaceCommit.
/// Used to skip re-encoding when the pixel data hasn't changed.
generation: u64,
/// CLOCK_MONOTONIC milliseconds captured at compositor commit time.
/// Used as the surface frame timestamp so the client sees the source's
/// presentation timing rather than the (jittery) encode-delivery clock.
timestamp_ms: u32,
}
struct SharedCompositor {
handle: CompositorHandle,
surfaces: HashMap<u16, CachedSurfaceInfo>,
/// Latest pixel snapshot per surface.
last_pixels: HashMap<u16, LastPixels>,
/// Per-surface timestamp of the last RequestFrame sent. Used to
/// throttle requests to at most one per 1 ms so frame callbacks
/// carry distinct `elapsed_ms` timestamps — video players (mpv)
/// use these to pace their presentation clock. Supports up to 1 kHz.
last_frame_request: HashMap<u16, Instant>,
created_at: Instant,
/// Monotonically increasing counter for pixel generations.
pixel_generation: u64,
/// Last time we sent blanket RequestFrame for all surfaces (including
/// those without subscribers). Throttled to prevent hot-looping when
/// apps commit at high rates without any client consuming frames.
last_blanket_frame_request: Instant,
/// Last dimensions sent to the compositor via `CompositorCommand::SurfaceResize`.
/// Used to dedup resize commands — the composited output size
/// (`info.width`/`info.height`) may differ from the requested size
/// when the Wayland client sets `xdg_geometry` (e.g. excluding a
/// title bar), so we compare against the actually-requested values.
last_configured_size: HashMap<u16, (u16, u16, u16)>,
/// Audio capture pipeline (PipeWire daemon → in-process libpipewire capture → Opus encode).
/// `None` when PipeWire is not available or `BLIT_AUDIO=0`.
#[cfg(target_os = "linux")]
audio_pipeline: Option<audio::AudioPipeline>,
/// Shared fan-out state for audio — subscribers, catch-up ring,
/// listener flag. Persistent across pipeline restarts so clients
/// stay subscribed even when the pipeline is restarted. Always present on Linux;
/// subscribe/unsubscribe succeeds even when the pipeline itself is
/// absent (frames just don't flow until it's back).
#[cfg(target_os = "linux")]
audio_broadcast: Arc<audio::AudioBroadcast>,
/// Compositor instance ID passed to `AudioPipeline::spawn()` so restarts
/// reuse the same audio runtime directory.
#[cfg(target_os = "linux")]
audio_session_id: u16,
/// When the last audio pipeline restart was attempted. Used to enforce a
/// cooldown so we don't spin on persistent failures.
#[cfg(target_os = "linux")]
last_audio_restart: Option<Instant>,
}
fn encode_rgba_to_png(pixels: &[u8], width: u32, height: u32) -> Vec<u8> {
let mut buf = Vec::new();
{
let expected = (width as usize) * (height as usize) * 4;
let actual = pixels.len();
if actual != expected {
// Size mismatch — return a 1×1 red pixel PNG rather than panicking.
let mut encoder = png::Encoder::new(&mut buf, 1, 1);
encoder.set_color(png::ColorType::Rgba);
encoder.set_depth(png::BitDepth::Eight);
let mut writer = encoder.write_header().unwrap();
writer.write_image_data(&[255, 0, 0, 255]).unwrap();
eprintln!(
"[capture] pixel buffer size mismatch: {width}x{height} expected {expected} got {actual}"
);
} else {
let mut encoder = png::Encoder::new(&mut buf, width, height);
encoder.set_color(png::ColorType::Rgba);
encoder.set_depth(png::BitDepth::Eight);
let mut writer = encoder.write_header().unwrap();
writer.write_image_data(pixels).unwrap();
}
}
buf
}
/// Encode RGBA pixels to AVIF. `quality` 0 = lossless, 1–100 = lossy.
fn encode_rgba_to_avif(pixels: &[u8], width: u32, height: u32, quality: u8) -> Vec<u8> {
let rgba: Vec<rgb::RGBA8> = pixels
.chunks_exact(4)
.map(|c| rgb::RGBA8::new(c[0], c[1], c[2], c[3]))
.collect();
let img = ravif::Img::new(&rgba[..], width as usize, height as usize);
let q = if quality == 0 { 100.0 } else { quality as f32 };
let encoder = ravif::Encoder::new()
.with_quality(q)
.with_alpha_quality(q)
.with_speed(6)
.with_alpha_color_mode(ravif::AlphaColorMode::UnassociatedClean)
.with_num_threads(None);
let result = encoder.encode_rgba(img).expect("AVIF encoding failed");
result.avif_file
}
/// Encode RGBA pixels to the requested capture format.
fn encode_capture(pixels: &[u8], width: u32, height: u32, format: u8, quality: u8) -> Vec<u8> {
match format {
CAPTURE_FORMAT_AVIF => encode_rgba_to_avif(pixels, width, height, quality),
_ => encode_rgba_to_png(pixels, width, height),
}
}
async fn request_surface_capture_with_timeout(
command_tx: std::sync::mpsc::Sender<CompositorCommand>,
surface_id: u16,
scale_120: u16,
timeout: Duration,
) -> Option<(u32, u32, Vec<u8>)> {
let (tx, rx) = std::sync::mpsc::sync_channel(1);
command_tx
.send(CompositorCommand::Capture {
surface_id,
scale_120,
reply: tx,
})
.ok()?;
// The compositor replies through a blocking std::sync::mpsc channel.
// Wait for it off the async runtime so this request never stalls the
// tokio worker thread or holds the Session mutex while blocked.
tokio::task::spawn_blocking(move || rx.recv_timeout(timeout))
.await
.ok()?
.ok()
.flatten()
}
/// Per-surface bookkeeping for an active subscription. Every field
/// defaults to "no-op" so a fresh `entry(sid).or_default()` is safe
/// even before any other state has been recorded.
#[derive(Default)]
struct SurfaceSubState {
/// Active encoder for this surface. `None` between encode jobs
/// while the encoder is temporarily owned by the spawn_blocking
/// task (see `encode_in_flight`) or before the first encode.
encoder: Option<SurfaceEncoder>,
/// Next tick this surface may send a frame (pacing deadline).
next_send_at: Option<Instant>,
/// Frames remaining in the post-subscribe burst window that
/// bypass time-based pacing so bandwidth estimates ramp up fast
/// on high-latency links.
burst_remaining: u8,
/// True while an encoder-creation spawn_blocking task is running
/// for this surface. Prevents dispatching a second creation in
/// parallel and (via the `needs_new_encoder` path) skips encode
/// dispatch until the creation task lands its result.
creation_in_flight: bool,
/// True while this surface's encoder is in an encode spawn_blocking
/// task. Prevents dispatching a parallel encode for the same
/// surface (the encoder has been moved into the task).
encode_in_flight: bool,
/// Set if the in-flight encoder was invalidated by a codec /
/// quality change (resubscribe) while encoding — the completion
/// handler must drop the stale encoder instead of reinserting it.
encoder_invalidated: bool,
/// Pixel generation that was last encoded; used to skip re-
/// encoding identical pixel data on subsequent ticks.
last_encoded_gen: Option<u64>,
/// Consecutive `nal_data=None` encodes. After too many, the
/// encoder is dropped so a fresh one is created on the next tick
/// (bounds runaway encoder-recreation loops).
nal_none_streak: u32,
/// When the streak last latched (hit the drop threshold). Auto-
/// clears after a backoff so a freshly-created encoder can retry
/// without needing a user-driven resize/resubscribe.
nal_none_latched_at: Option<Instant>,
/// Per-surface codec support override from C2S_SURFACE_SUBSCRIBE
/// (bitmask of CODEC_SUPPORT_*). 0 = defer to client-wide
/// `surface_codec_support`.
codec_override: u8,
/// Per-surface quality override. `None` = use server default.
quality_override: Option<SurfaceQuality>,
}
struct ClientState {
tx: mpsc::UnboundedSender<Vec<u8>>,
outbox_queued_frames: Arc<AtomicUsize>,
outbox_queued_bytes: Arc<AtomicUsize>,
/// Dedicated channel for audio frames. The writer task selects on this
/// with higher priority than the main outbox so audio is never starved
/// by large video/terminal messages.
audio_tx: mpsc::UnboundedSender<Vec<u8>>,
lead: Option<u16>,
subscriptions: HashSet<u16>,
/// Active surface subscriptions for this client.
surface_subscriptions: HashSet<u16>,
/// Whether this client is subscribed to audio frames.
audio_subscribed: bool,
/// Per-client audio bitrate preference in kbps from C2S_AUDIO_SUBSCRIBE.
/// 0 means use the server/env default.
#[cfg(target_os = "linux")]
audio_bitrate_kbps: u16,
view_sizes: HashMap<u16, (u16, u16)>,
scroll_offsets: HashMap<u16, usize>,
scroll_caches: HashMap<u16, FrameState>,
last_sent: HashMap<u16, FrameState>,
preview_next_send_at: HashMap<u16, Instant>,
/// EWMA RTT estimate in milliseconds.
rtt_ms: f32,
/// Minimum-path RTT estimate in milliseconds, excluding queue growth.
min_rtt_ms: f32,
/// Client's measured display refresh rate (fps), reported via C2S_DISPLAY_RATE.
display_fps: f32,
/// EWMA of delivered payload rate in bytes/sec.
delivery_bps: f32,
/// EWMA of actual ACKed goodput in bytes/sec, based on ACK cadence rather than RTT.
goodput_bps: f32,
/// EWMA of absolute goodput sample-to-sample jitter in bytes/sec.
goodput_jitter_bps: f32,
/// Decaying peak goodput jitter in bytes/sec.
max_goodput_jitter_bps: f32,
/// Last sampled ACK goodput for jitter estimation.
last_goodput_sample_bps: f32,
/// EWMA of acknowledged frame payload size in bytes.
avg_frame_bytes: f32,
/// EWMA of acknowledged lead/paced frame payload size in bytes.
avg_paced_frame_bytes: f32,
/// EWMA of acknowledged preview/unpaced frame payload size in bytes.
avg_preview_frame_bytes: f32,
/// EWMA of surface (video) frame payload size in bytes. Tracked
/// separately from terminal frame sizes so surface pacing uses
/// `goodput_bps / avg_surface_frame_bytes` without polluting
/// terminal congestion control estimates.
avg_surface_frame_bytes: f32,
/// Payload bytes currently in flight (sent, not yet ACKed).
inflight_bytes: usize,
/// Oldest in-flight frame first; ACKs arrive in order.
inflight_frames: VecDeque<InFlightFrame>,
/// Earliest time the next visual update should be sent for smooth pacing.
next_send_at: Instant,
/// Temporary additive window growth used to probe for more throughput after
/// a conservative backoff. Decays when queue delay grows.
probe_frames: f32,
/// Diagnostics.
frames_sent: u32,
acks_recv: u32,
acked_bytes_since_log: usize,
browser_backlog_frames: u16,
browser_ack_ahead_frames: u16,
browser_apply_ms: f32,
last_metrics_update: Instant,
last_log: Instant,
/// Throttle timestamp for "[surface-gate] blocked" diagnostic logs.
last_window_blocked_log: Instant,
/// Throttle timestamp for "[encode-skip]" diagnostic logs.
last_skip_log: Instant,
/// Counters for silent encode-skip paths, reset each pacing log tick.
skip_same_gen_count: u32,
skip_in_flight_count: u32,
skip_pacing_count: u32,
skip_vulkan_await_count: u32,
/// Client had no subscriptions when encode pass ran.
skip_no_subs_count: u32,
/// Client not subscribed to a given sid in pixel_snapshot.
skip_not_subbed_count: u32,
/// last_pixels entry missing / dimensions mismatched pixel_snapshot.
skip_last_pixels_mismatch_count: u32,
/// Iterations through pixel_snapshot for this client (sanity check).
encode_loop_iters: u32,
goodput_window_bytes: usize,
goodput_window_start: Instant,
/// Per-surface encode/pacing/override state. Holds every piece of
/// bookkeeping the encode loop maintains between frames for a
/// subscribed surface. Entries are created lazily via
/// `entry(sid).or_default()` on first touch and dropped wholesale
/// on UNSUBSCRIBE / SurfaceDestroyed.
surface_subs: HashMap<u16, SurfaceSubState>,
surface_needs_keyframe: bool,
/// Surfaces that use Vulkan Video encoding in the compositor rather than
/// a local SurfaceEncoder. Maps surface_id → (encoder_name, codec_flag).
vulkan_video_surfaces: HashMap<u16, (&'static str, u8)>,
/// Surface frames in flight — separate from terminal inflight so surface
/// ACKs feed shared RTT / goodput without corrupting terminal frame-size
/// averages or probe_frames.
surface_inflight_frames: VecDeque<InFlightFrame>,
/// Per-client desired surface sizes (surface_id → (width, height, scale_120, codec_support)).
/// Mirrors `view_sizes` for PTYs: the server mediates across all clients
/// and picks min(width), min(height), max(scale).
/// `scale_120` is the DPR in 1/120th units (Wayland convention): 240 = 2×.
surface_view_sizes: HashMap<u16, (u16, u16, u16)>,
/// Intersection of codec support across all surfaces for this client.
/// Used to pick an encoder the client can decode. 0 = accept anything.
surface_codec_support: u8,
/// Evdev keycodes currently held down by this client on compositor
/// surfaces. On disconnect we send synthetic key-up events for each
/// so modifiers don't stay stuck and keys don't auto-repeat forever.
pressed_surface_keys: HashSet<u32>,
}
struct InFlightFrame {
sent_at: Instant,
bytes: usize,
paced: bool,
}
/// Frames to keep in flight: enough to cover one RTT at the client's reported
/// display rate. High-latency links need many frames in flight to avoid
/// devolving into stop-and-wait.
fn frame_window(rtt_ms: f32, display_fps: f32) -> usize {
let frame_ms = 1_000.0 / display_fps.max(1.0);
let base_frames = (rtt_ms / frame_ms).ceil().max(0.0) as usize;
let slack_frames = ((base_frames as f32) * 0.125).ceil() as usize + 2;
base_frames.saturating_add(slack_frames).max(2)
}
fn path_rtt_ms(client: &ClientState) -> f32 {
if client.min_rtt_ms > 0.0 {
client.min_rtt_ms
} else {
client.rtt_ms
}
}
fn display_need_bps(client: &ClientState) -> f32 {
client.avg_paced_frame_bytes.max(256.0) * client.display_fps.max(1.0)
}
fn effective_rtt_ms(client: &ClientState) -> f32 {
let path_rtt = path_rtt_ms(client);
let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
let queue_allowance = frame_ms
* if throughput_limited(client) {
4.0
} else {
12.0
};
client.rtt_ms.clamp(path_rtt, path_rtt + queue_allowance)
}
fn window_rtt_ms(client: &ClientState) -> f32 {
let effective = effective_rtt_ms(client);
if !throughput_limited(client) {
effective
} else {
client.rtt_ms.clamp(effective, effective * 2.0)
}
}
fn target_frame_window(client: &ClientState) -> usize {
let window_fps = if throughput_limited(client) {
pacing_fps(client)
} else {
browser_pacing_fps(client)
};
frame_window(window_rtt_ms(client), window_fps)
.saturating_add(client.probe_frames.round().max(0.0) as usize)
}
fn base_queue_ms(client: &ClientState) -> f32 {
let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
frame_ms * if throughput_limited(client) { 2.0 } else { 8.0 }
}
fn target_queue_ms(client: &ClientState) -> f32 {
let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
let probe_scale = if throughput_limited(client) {
0.25
} else {
1.0
};
base_queue_ms(client) + client.probe_frames.max(0.0) * frame_ms * probe_scale
}
fn browser_ready(client: &ClientState) -> bool {
client.browser_ack_ahead_frames <= 1
&& client.browser_apply_ms <= 1.0
&& !outbox_backpressured(client)
}
fn bandwidth_floor_bps(client: &ClientState) -> f32 {
let browser_ready = browser_ready(client);
let backlog_scale = match client.browser_backlog_frames {
0..=2 => 0.9,
3..=8 => 0.8,
_ => 0.65,
};
let penalty = client
.goodput_jitter_bps
.max(client.max_goodput_jitter_bps * 0.5)
.min(client.goodput_bps * if browser_ready { 0.75 } else { 0.9 });
let goodput_floor = (client.goodput_bps - penalty)
.max(client.goodput_bps * if browser_ready { 0.35 } else { 0.2 });
// On a browser-ready path, the per-frame delivery estimate is already
// end-to-end and reacts much faster than ACK-window goodput. Halving it
// leaves large-frame local links chronically underpaced.
let delivery_floor = client.delivery_bps * if browser_ready { 1.0 } else { 0.5 };
let recent_sample_floor = if browser_ready && client.last_goodput_sample_bps > 0.0 {
client.last_goodput_sample_bps * backlog_scale
} else {
0.0
};
goodput_floor.max(recent_sample_floor).max(delivery_floor)
}
fn pacing_fps(client: &ClientState) -> f32 {
let frame_bytes = client.avg_paced_frame_bytes.max(256.0);
let sustainable = bandwidth_floor_bps(client) / frame_bytes;
sustainable.min(browser_pacing_fps(client))
}
fn throughput_limited(client: &ClientState) -> bool {
let floor = bandwidth_floor_bps(client);
// Consider total demand: lead at cadence rate plus previews at their cap.
// The old check (pacing_fps < cadence * 0.9) only saw lead bandwidth,
// which is often tiny, so previews could starve the lead undetected.
let lead_bps = client.avg_paced_frame_bytes.max(256.0) * browser_pacing_fps(client);
let preview_bps = client.avg_preview_frame_bytes.max(256.0) * client.display_fps.max(1.0);
(lead_bps + preview_bps) > floor * 0.9
}
fn browser_pacing_fps(client: &ClientState) -> f32 {
let mut fps = client.display_fps.max(1.0);
// Backlog and ack-ahead are direct signals from the browser about
// whether it's keeping up. No predictive apply-time bound — it
// consistently underestimates capacity and causes 30fps death spirals.
//
// The backoff is steep: at the block threshold (backlog>8) we've
// already dropped to display_fps/4. A gentler schedule (4/backlog)
// held 48fps at backlog=10 for software-encoded 1080p, which is
// faster than the browser can decode → backlog never drains, the
// hard block stays latched, and encoding stalls entirely.
//
// Trigger threshold (backlog > 4) gives a few frames of transient
// headroom before backoff engages — at 120 Hz, a 30 fps source naturally
// queues 1-2 frames during decoder hiccups, and triggering backoff there
// chops the rate just to absorb normal jitter.
let backlog = client.browser_backlog_frames as f32;
if backlog > 4.0 {
fps = fps.min(fps * (2.0 / backlog));
}
if client.browser_ack_ahead_frames > 4 {
fps = fps.min(client.display_fps.max(1.0) * 0.5);
}
if client.browser_ack_ahead_frames > 8 {
fps = fps.min(client.display_fps.max(1.0) * 0.25);
}
fps.max(1.0)
}
fn browser_backlog_blocked(client: &ClientState) -> bool {
client.browser_backlog_frames > 8
}
fn byte_budget_for(client: &ClientState, budget_ms: f32) -> usize {
let budget_bps = if throughput_limited(client) {
bandwidth_floor_bps(client)
} else {
client.goodput_bps.max(bandwidth_floor_bps(client))
};
let bytes = budget_bps * budget_ms.max(1.0) / 1_000.0;
bytes.ceil().max(client.avg_frame_bytes.max(256.0)) as usize
}
fn target_byte_window(client: &ClientState) -> usize {
let budget = byte_budget_for(client, path_rtt_ms(client) + target_queue_ms(client));
let frame_bytes = client.avg_paced_frame_bytes.max(256.0).ceil() as usize;
let target_frames = target_frame_window(client);
let pipeline_bytes = frame_bytes.saturating_mul(target_frames);
// For small pipelines (e.g. idle terminals with 1KB frames), allow the
// full frame window worth of bytes so we pipeline across the RTT instead
// of stop-and-wait. For large pipelines (e.g. 50KB frames × 5 frames =
// 250KB), the budget (BDP-based) is the binding constraint; fall back to
// a one-frame floor so we don't pile up many RTTs worth of large frames.
const PIPELINE_FLOOR_LIMIT: usize = 32_768; // 32 KB
let floor = if pipeline_bytes <= PIPELINE_FLOOR_LIMIT {
pipeline_bytes
} else {
frame_bytes // one-frame floor for large pipelines
};
budget.max(floor)
}
fn send_interval(client: &ClientState) -> Duration {
Duration::from_secs_f64(1.0 / browser_pacing_fps(client).max(1.0) as f64)
}
fn preview_fps(client: &ClientState) -> f32 {
let mut fps = client.display_fps.max(1.0);
if client.lead.is_some() && throughput_limited(client) {
// Only budget preview bandwidth when the link is actually saturated.
// Without this, large preview frames (e.g. 12 KB) at 30 fps consume
// 360 KB/s, starving the lead even when lead frames are tiny.
// On fast links (localhost, LAN), previews run at display_fps.
let avail = bandwidth_floor_bps(client);
let lead_bps = client.avg_paced_frame_bytes.max(256.0) * browser_pacing_fps(client);
let preview_budget = (avail - lead_bps).max(avail * 0.25).max(0.0);
let bw_cap = preview_budget / client.avg_preview_frame_bytes.max(256.0);
fps = fps.min(bw_cap.max(1.0));
}
fps.max(1.0)
}
fn preview_send_interval(client: &ClientState) -> Duration {
Duration::from_secs_f64(1.0 / preview_fps(client) as f64)
}
/// Surface frame rate: delegates to `browser_pacing_fps` so video backs off
/// when the client reports it can't keep up (backlog / ack-ahead growth).
/// Continuing at display_fps when the browser is behind just piles up video
/// it will drop to catch up, which the user perceives as droppy playback.
fn surface_pacing_fps(client: &ClientState) -> f32 {
browser_pacing_fps(client)
}
fn surface_send_interval(client: &ClientState) -> Duration {
Duration::from_secs_f64(1.0 / surface_pacing_fps(client).max(1.0) as f64)
}
/// Emit a pacing-metrics line for this client if 10s have elapsed since
/// the last one. Called both from the ACK handler and from `tick()` so
/// an idle client (no ACK traffic) still gets periodic metrics.
fn maybe_log_pacing_metrics(sess: &mut Session, client_id: u64, verbose: bool) {
let Some(c) = sess.clients.get_mut(&client_id) else {
return;
};
if c.last_log.elapsed().as_secs_f32() < 10.0 {
return;
}
let log_elapsed = c.last_log.elapsed().as_secs_f32().max(1.0e-3);
let paced_fps = pacing_fps(c);
let display_need_bps_v = display_need_bps(c);
let surface_fps = surface_pacing_fps(c);
let frames_sent = c.frames_sent;
let acks_recv = c.acks_recv;
let rtt_ms = c.rtt_ms;
let min_rtt_ms = path_rtt_ms(c);
let eff_rtt_ms = window_rtt_ms(c);
let inflight_bytes = c.inflight_bytes;
let delivery_bps = c.delivery_bps;
let goodput_ewma_bps = c.goodput_bps;
let goodput_jitter_bps = c.goodput_jitter_bps;
let max_goodput_jitter_bps = c.max_goodput_jitter_bps;
let avg_frame_bytes = c.avg_frame_bytes;
let avg_paced_frame_bytes = c.avg_paced_frame_bytes;
let avg_preview_frame_bytes = c.avg_preview_frame_bytes;
let display_fps = c.display_fps;
let probe_frames = c.probe_frames;
let goodput_bps = c.acked_bytes_since_log as f32 / log_elapsed;
let window_frames = target_frame_window(c);
let window_bytes = target_byte_window(c);
let outbox_frames = outbox_queued_frames(c);
let browser_backlog_frames = c.browser_backlog_frames;
let browser_ack_ahead_frames = c.browser_ack_ahead_frames;
let browser_apply_ms = c.browser_apply_ms;
let avg_surface_frame_bytes = c.avg_surface_frame_bytes;
let skip_same_gen = c.skip_same_gen_count;
let skip_in_flight = c.skip_in_flight_count;
let skip_pacing = c.skip_pacing_count;
let skip_vk_await = c.skip_vulkan_await_count;
let skip_no_subs = c.skip_no_subs_count;
let skip_not_subbed = c.skip_not_subbed_count;
let skip_mismatch = c.skip_last_pixels_mismatch_count;
let loop_iters = c.encode_loop_iters;
let own_subs: usize = c.surface_subscriptions.len();
let vk_surfs = c.vulkan_video_surfaces.len();
let in_flight_set_len = c
.surface_subs
.values()
.filter(|s| s.encode_in_flight)
.count();
let surface_burst: u8 = c
.surface_subs
.values()
.map(|s| s.burst_remaining)
.max()
.unwrap_or(0);
c.frames_sent = 0;
c.acks_recv = 0;
c.acked_bytes_since_log = 0;
c.skip_same_gen_count = 0;
c.skip_in_flight_count = 0;
c.skip_pacing_count = 0;
c.skip_vulkan_await_count = 0;
c.skip_no_subs_count = 0;
c.skip_not_subbed_count = 0;
c.skip_last_pixels_mismatch_count = 0;
c.encode_loop_iters = 0;
c.last_log = Instant::now();
if verbose {
let surf_info = sess.compositor.as_ref().map(|cs| {
let surfaces = cs.surfaces.len();
let pending = 0usize;
let subs: usize = sess
.clients
.values()
.map(|c| c.surface_subscriptions.len())
.sum();
(surfaces, pending, subs)
});
let (surf_count, surf_pending, surf_subs) = surf_info.unwrap_or((0, 0, 0));
eprintln!(
"client {client_id}: sent={frames_sent} acks={acks_recv} rtt={rtt_ms:.0}ms min_rtt={min_rtt_ms:.0}ms eff_rtt={eff_rtt_ms:.0}ms window={window_frames}f/{window_bytes}B probe={probe_frames:.0}f inflight={inflight_bytes}B outbox={outbox_frames}f goodput={goodput_bps:.0}B/s goodput_ewma={goodput_ewma_bps:.0}B/s jitter={goodput_jitter_bps:.0}/{max_goodput_jitter_bps:.0}B/s rate={delivery_bps:.0}B/s avg_frame={avg_frame_bytes:.0}B lead_frame={avg_paced_frame_bytes:.0}B preview_frame={avg_preview_frame_bytes:.0}B need={display_need_bps_v:.0}B/s display_fps={display_fps:.0} paced_fps={paced_fps:.0} surface_fps={surface_fps:.0} surface_frame={avg_surface_frame_bytes:.0}B backlog={browser_backlog_frames} ack_ahead={browser_ack_ahead_frames} apply={browser_apply_ms:.1}ms | tick_fires={} tick_snaps={} | surfaces={surf_count} subs={surf_subs} own_subs={own_subs} pending_req={surf_pending} commits={} encodes={} enc_bytes={} surf_sent={} px_empty_ticks={} px_snap_len={} loop_iters={loop_iters} skip_same_gen={skip_same_gen} skip_in_flight={skip_in_flight} skip_pacing={skip_pacing} skip_vk_await={skip_vk_await} skip_no_subs={skip_no_subs} skip_not_subbed={skip_not_subbed} skip_mismatch={skip_mismatch} vk_surfs={vk_surfs} enc_in_flight_set={in_flight_set_len} burst={surface_burst}",
sess.tick_fires,
sess.tick_snaps,
sess.surface_commits,
sess.surface_encodes,
sess.surface_encode_bytes,
sess.surface_frames_sent,
sess.ticks_pixel_snapshot_empty,
sess.pixel_snapshot_len,
);
}
sess.tick_fires = 0;
sess.tick_snaps = 0;
sess.surface_commits = 0;
sess.surface_encodes = 0;
sess.surface_encode_bytes = 0;
sess.surface_frames_sent = 0;
sess.ticks_pixel_snapshot_empty = 0;
}
fn advance_deadline(deadline: &mut Instant, now: Instant, interval: Duration) {
let scheduled = deadline.checked_add(interval).unwrap_or(now + interval);
*deadline = if scheduled + interval < now {
now + interval
} else {
scheduled
};
}
fn should_snapshot_pty(dirty: bool, needful: bool, synced_output: bool) -> bool {
dirty && needful && !synced_output
}
fn enqueue_ready_frame(queue: &mut VecDeque<FrameState>, frame: FrameState) -> bool {
if queue.len() >= READY_FRAME_QUEUE_CAP {
return false;
}
queue.push_back(frame);
true
}
fn pty_has_visual_update(pty: &Pty) -> bool {
pty.dirty || !pty.ready_frames.is_empty() || !pty.byte_rx.is_empty()
}
/// Find the first `\x1b[?2026l` in `bytes`, handling sequences that span
/// the `prefix`/`bytes` boundary. Uses SIMD-accelerated memchr for the
/// initial ESC scan.
fn find_sync_output_end(prefix: &[u8], bytes: &[u8]) -> Option<usize> {
if bytes.is_empty() {
return None;
}
let needle = SYNC_OUTPUT_END;
let nlen = needle.len();
// Check for a match straddling the prefix/bytes boundary.
if !prefix.is_empty() {
let tail = if prefix.len() >= nlen - 1 {
&prefix[prefix.len() - (nlen - 1)..]
} else {
prefix
};
let combined_len = tail.len() + bytes.len().min(nlen);
if combined_len >= nlen {
// Small stack buffer to check the boundary region.
let mut buf = [0u8; 32]; // SYNC_OUTPUT_END is 8 bytes, so 32 is plenty
let blen = combined_len.min(buf.len());
let tlen = tail.len().min(blen);
buf[..tlen].copy_from_slice(&tail[..tlen]);
let rest = (blen - tlen).min(bytes.len());
buf[tlen..tlen + rest].copy_from_slice(&bytes[..rest]);
for i in 0..=(blen.saturating_sub(nlen)) {
if &buf[i..i + nlen] == needle {
let end_in_bytes = (i + nlen).saturating_sub(tail.len());
if end_in_bytes > 0 && end_in_bytes <= bytes.len() {
return Some(end_in_bytes);
}
}
}
}
}
// SIMD-scan for ESC (0x1b) then verify the full sequence.
let mut offset = 0;
while let Some(pos) = memchr::memchr(0x1b, &bytes[offset..]) {
let abs = offset + pos;
if abs + nlen <= bytes.len() && &bytes[abs..abs + nlen] == needle {
return Some(abs + nlen);
}
offset = abs + 1;
}
None
}
fn update_sync_scan_tail(tail: &mut Vec<u8>, bytes: &[u8]) {
if bytes.is_empty() {
return;
}
tail.extend_from_slice(bytes);
let keep = SYNC_OUTPUT_END.len().saturating_sub(1);
if tail.len() > keep {
let drop = tail.len() - keep;
tail.drain(..drop);
}
}
fn preview_deadline(client: &ClientState, pid: u16, now: Instant) -> Instant {
client
.preview_next_send_at
.get(&pid)
.copied()
.unwrap_or(now)
}
fn client_has_due_preview(sess: &Session, client: &ClientState, now: Instant) -> bool {
if client.lead.is_none() {
return false;
}
client.subscriptions.iter().copied().any(|pid| {
Some(pid) != client.lead
&& preview_deadline(client, pid, now) <= now
&& sess
.ptys
.get(&pid)
.map(pty_has_visual_update)
.unwrap_or(false)
})
}
fn outbox_queued_frames(client: &ClientState) -> usize {
client.outbox_queued_frames.load(Ordering::Relaxed)
}
fn outbox_queued_bytes(client: &ClientState) -> usize {
client.outbox_queued_bytes.load(Ordering::Relaxed)
}
fn outbox_backpressured(client: &ClientState) -> bool {
// Always allow at least one frame queued, even if it exceeds the byte
// soft limit. Large keyframes from software encoders can be larger than
// OUTBOX_SOFT_QUEUE_LIMIT_BYTES; treating the first queued frame as
// backpressure would permanently close surface_window_open and deadlock
// encoding (the one queued frame cannot drain until the sender task
// flushes it, but the sender was waiting for a new frame that we
// refuse to produce — deadlock).
let frames = outbox_queued_frames(client);
if frames >= OUTBOX_SOFT_QUEUE_LIMIT_FRAMES {
return true;
}
frames >= 2 && outbox_queued_bytes(client) >= OUTBOX_SOFT_QUEUE_LIMIT_BYTES
}
fn mark_outbox_drained(
queued_frames: &Arc<AtomicUsize>,
queued_bytes: &Arc<AtomicUsize>,
bytes: usize,
) {
let _ = queued_frames.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
Some(value.saturating_sub(1))
});
let _ = queued_bytes.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
Some(value.saturating_sub(bytes))
});
}
fn send_outbox_tracked(
tx: &mpsc::UnboundedSender<Vec<u8>>,
queued_frames: &Arc<AtomicUsize>,
queued_bytes: &Arc<AtomicUsize>,
msg: Vec<u8>,
) -> Result<(), mpsc::error::SendError<Vec<u8>>> {
let bytes = msg.len();
tx.send(msg)?;
queued_frames.fetch_add(1, Ordering::Relaxed);
queued_bytes.fetch_add(bytes, Ordering::Relaxed);
Ok(())
}
fn send_outbox(client: &ClientState, msg: Vec<u8>) -> Result<(), mpsc::error::SendError<Vec<u8>>> {
send_outbox_tracked(
&client.tx,
&client.outbox_queued_frames,
&client.outbox_queued_bytes,
msg,
)
}
fn can_send_preview(client: &ClientState, pid: u16, now: Instant) -> bool {
window_open(client) && now >= preview_deadline(client, pid, now)
}
fn record_preview_send(client: &mut ClientState, pid: u16, now: Instant) {
let mut deadline = client
.preview_next_send_at
.get(&pid)
.copied()
.unwrap_or(now);
advance_deadline(&mut deadline, now, preview_send_interval(client));
client.preview_next_send_at.insert(pid, deadline);
}
fn window_open(client: &ClientState) -> bool {
!browser_backlog_blocked(client)
&& !outbox_backpressured(client)
&& client.inflight_frames.len() < target_frame_window(client)
&& client.inflight_bytes < target_byte_window(client)
}
/// Surface send gate: outbox backpressure only. Rate is governed by
/// `surface_send_interval`; per-surface encode concurrency by the
/// `encode_in_flight` flag on `SurfaceSubState`.
fn surface_window_open(client: &ClientState) -> bool {
!outbox_backpressured(client)
}
fn lead_window_open(client: &ClientState, reserve_preview_slot: bool) -> bool {
if !reserve_preview_slot || client.lead.is_none() {
return window_open(client);
}
if browser_backlog_blocked(client) || outbox_backpressured(client) {
return false;
}
let target_frames = target_frame_window(client);
let reserve_frames = PREVIEW_FRAME_RESERVE.min(target_frames.saturating_sub(1));
let frame_limit = target_frames.saturating_sub(reserve_frames).max(1);
let reserve_bytes = client.avg_preview_frame_bytes.max(256.0).ceil() as usize;
let byte_limit = target_byte_window(client)
.saturating_sub(reserve_bytes)
.max(client.avg_paced_frame_bytes.max(256.0).ceil() as usize);
client.inflight_frames.len() < frame_limit && client.inflight_bytes < byte_limit
}
fn can_send_frame(client: &ClientState, now: Instant, reserve_preview_slot: bool) -> bool {
lead_window_open(client, reserve_preview_slot) && now >= client.next_send_at
}
fn record_send(client: &mut ClientState, bytes: usize, now: Instant, paced: bool) {
client.inflight_bytes += bytes;
client.inflight_frames.push_back(InFlightFrame {
sent_at: now,
bytes,
paced,
});
if paced {
let interval = send_interval(client);
advance_deadline(&mut client.next_send_at, now, interval);
}
}
fn ewma_with_direction(old: f32, sample: f32, rise_alpha: f32, fall_alpha: f32) -> f32 {
let alpha = if sample > old { rise_alpha } else { fall_alpha };
old * (1.0 - alpha) + sample * alpha
}
fn window_saturated(client: &ClientState, inflight_frames: usize, inflight_bytes: usize) -> bool {
let target_frames = target_frame_window(client);
let target_bytes = target_byte_window(client);
inflight_frames.saturating_mul(10) >= target_frames.saturating_mul(9)
|| inflight_bytes.saturating_mul(10) >= target_bytes.saturating_mul(9)
}
fn record_ack(client: &mut ClientState) {
if let Some(frame) = client.inflight_frames.pop_front() {
let prev_inflight_frames = client.inflight_frames.len() + 1;
let prev_inflight_bytes = client.inflight_bytes;
client.inflight_bytes = client.inflight_bytes.saturating_sub(frame.bytes);
client.acked_bytes_since_log = client.acked_bytes_since_log.saturating_add(frame.bytes);
let sample_ms = frame.sent_at.elapsed().as_secs_f32() * 1_000.0;
client.rtt_ms = ewma_with_direction(client.rtt_ms, sample_ms, 0.125, 0.25);
if client.min_rtt_ms > 0.0 {
// Only update downward: min_rtt tracks the unloaded path RTT and
// must not drift upward during congestion (queued RTT ≠ path RTT).
client.min_rtt_ms = client.min_rtt_ms.min(sample_ms);
} else {
client.min_rtt_ms = sample_ms;
}
client.min_rtt_ms = client.min_rtt_ms.max(0.5);
let sample_bps = frame.bytes as f32 / sample_ms.max(1.0e-3) * 1_000.0;
client.delivery_bps = ewma_with_direction(client.delivery_bps, sample_bps, 0.5, 0.125);
client.avg_frame_bytes =
ewma_with_direction(client.avg_frame_bytes, frame.bytes as f32, 0.5, 0.125);
if frame.paced {
client.avg_paced_frame_bytes =
ewma_with_direction(client.avg_paced_frame_bytes, frame.bytes as f32, 0.5, 0.125);
} else {
client.avg_preview_frame_bytes = ewma_with_direction(
client.avg_preview_frame_bytes,
frame.bytes as f32,
0.5,
0.125,
);
}
let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
let path_rtt = path_rtt_ms(client);
let likely_window_limited =
window_saturated(client, prev_inflight_frames, prev_inflight_bytes);
client.goodput_window_bytes = client.goodput_window_bytes.saturating_add(frame.bytes);
let now = Instant::now();
let goodput_elapsed = now
.duration_since(client.goodput_window_start)
.as_secs_f32();
if goodput_elapsed >= 0.02 {
let sample_goodput = client.goodput_window_bytes as f32 / goodput_elapsed.max(1.0e-3);
if likely_window_limited || client.browser_backlog_frames > 0 {
let prev_goodput_sample = if client.last_goodput_sample_bps > 0.0 {
client.last_goodput_sample_bps
} else {
sample_goodput
};
let jitter_sample = (sample_goodput - prev_goodput_sample).abs();
client.goodput_bps =
ewma_with_direction(client.goodput_bps, sample_goodput, 0.5, 0.125);
// Only update jitter from windows with at least 2 frames.
// Single-frame windows are pure measurement noise (0 or 1
// frame per 25 ms is a Bernoulli trial, not a congestion
// signal) and inflate jitter_bps, which in turn depresses
// bandwidth_floor_bps and causes pacing to stall.
let min_reliable = (client.avg_paced_frame_bytes.max(256.0) * 2.0) as usize;
if client.goodput_window_bytes >= min_reliable {
client.goodput_jitter_bps =
ewma_with_direction(client.goodput_jitter_bps, jitter_sample, 0.5, 0.125);
let jitter_decay = if browser_ready(client) && sample_ms < path_rtt * 3.0 {
0.90
} else {
0.98
};
client.max_goodput_jitter_bps =
(client.max_goodput_jitter_bps * jitter_decay).max(jitter_sample);
// Cap jitter at 45% of goodput so jitter_ratio can never
// exceed 0.45 from measurement noise alone. Real congestion
// will still drive goodput_bps down and widen the window.
client.max_goodput_jitter_bps =
client.max_goodput_jitter_bps.min(client.goodput_bps * 0.45);
} else {
// Thin sample: gently decay jitter rather than updating it.
client.goodput_jitter_bps *= 0.9;
client.max_goodput_jitter_bps *= 0.95;
}
// Sticky-high: never let last_goodput_sample_bps drop abruptly.
// A sudden drop (e.g. 1-frame window following a 2-frame window)
// inflates jitter_sample on the next cycle, collapsing probe_frames.
client.last_goodput_sample_bps =
(client.last_goodput_sample_bps * 0.99).max(sample_goodput);
} else {
// When the path is underfilled, ACK cadence mostly measures our
// own pacing rather than network capacity. Use a fall alpha
// proportional to estimation error: when the estimate is 10x+
// the sample, converge aggressively; when close, stay gentle.
let ratio = client.goodput_bps / sample_goodput.max(1.0);
let fall_alpha = if ratio > 10.0 {
0.5
} else if ratio > 3.0 {
0.25
} else {
0.03
};
client.goodput_bps =
ewma_with_direction(client.goodput_bps, sample_goodput, 0.5, fall_alpha);
client.goodput_jitter_bps *= 0.5;
client.max_goodput_jitter_bps *= 0.9;
client.last_goodput_sample_bps =
(client.last_goodput_sample_bps * 0.99).max(sample_goodput);
}
client.goodput_window_bytes = 0;
client.goodput_window_start = now;
}
let queue_baseline_ms = if throughput_limited(client) {
window_rtt_ms(client)
} else {
path_rtt
};
let queue_delay_ms = (sample_ms - queue_baseline_ms).max(0.0);
let max_probe_frames = (browser_pacing_fps(client) * 0.125).max(4.0);
let jitter_ratio = client.max_goodput_jitter_bps / client.goodput_bps.max(1.0);
let low_delay_frames = if throughput_limited(client) { 2.0 } else { 8.0 };
let high_delay_frames = if throughput_limited(client) {
4.0
} else {
12.0
};
if likely_window_limited
&& queue_delay_ms <= frame_ms * low_delay_frames
&& jitter_ratio < 0.25
{
client.probe_frames = (client.probe_frames + 1.0).min(max_probe_frames);
} else if !likely_window_limited
&& browser_ready(client)
&& queue_delay_ms <= frame_ms * 2.0
&& jitter_ratio < 0.25
{
client.probe_frames = (client.probe_frames + 0.25).min(max_probe_frames * 0.5);
} else if queue_delay_ms > frame_ms * high_delay_frames || jitter_ratio > 0.5 {
client.probe_frames = (client.probe_frames * 0.5).max(1.0);
} else if queue_delay_ms > frame_ms * 2.0 || !browser_ready(client) {
client.probe_frames = (client.probe_frames - 0.5).max(0.0);
}
} else {
client.inflight_bytes = 0;
}
}
/// Process a surface ACK. Feeds delivery_bps and goodput estimates (same
/// pipe) from the surface inflight queue. Does NOT update rtt_ms / min_rtt_ms
/// — surface frames are large and their wall-clock delivery time is dominated
/// by serialization and wire transfer, not network latency. Feeding those
/// samples into the shared RTT inflates it by orders of magnitude and
/// destabilises terminal pacing and congestion control.
fn record_surface_ack(client: &mut ClientState) {
if let Some(frame) = client.surface_inflight_frames.pop_front() {
client.acked_bytes_since_log = client.acked_bytes_since_log.saturating_add(frame.bytes);
let sample_ms = frame.sent_at.elapsed().as_secs_f32() * 1_000.0;
// Shared delivery rate (bandwidth, not latency — safe to update).
let sample_bps = frame.bytes as f32 / sample_ms.max(1.0e-3) * 1_000.0;
client.delivery_bps = ewma_with_direction(client.delivery_bps, sample_bps, 0.5, 0.125);
// Shared goodput window — accumulate bytes, flush periodically.
// Surface traffic at display_fps is sustained, so always use the
// window-limited EWMA parameters (rise 0.5, fall 0.125). No
// jitter tracking — jitter is a terminal congestion-control signal
// and large keyframe/P-frame variance would poison it.
client.goodput_window_bytes = client.goodput_window_bytes.saturating_add(frame.bytes);
let now = Instant::now();
let goodput_elapsed = now
.duration_since(client.goodput_window_start)
.as_secs_f32();
if goodput_elapsed >= 0.02 {
let sample_goodput = client.goodput_window_bytes as f32 / goodput_elapsed.max(1.0e-3);
client.goodput_bps =
ewma_with_direction(client.goodput_bps, sample_goodput, 0.5, 0.125);
client.last_goodput_sample_bps =
(client.last_goodput_sample_bps * 0.99).max(sample_goodput);
client.goodput_window_bytes = 0;
client.goodput_window_start = now;
}
}
}
fn reset_inflight(client: &mut ClientState) {
client.inflight_bytes = 0;
client.inflight_frames.clear();
client.next_send_at = Instant::now();
client.browser_backlog_frames = 0;
client.browser_ack_ahead_frames = 0;
}
fn is_unset_view_size(rows: u16, cols: u16) -> bool {
rows == 0 && cols == 0
}
fn subscribe_client_to(client: &mut ClientState, pty_id: u16) {
if client.subscriptions.insert(pty_id) {
client.last_sent.remove(&pty_id);
client.preview_next_send_at.remove(&pty_id);
}
}
fn unsubscribe_client_from(client: &mut ClientState, pty_id: u16) -> bool {
let removed_sub = client.subscriptions.remove(&pty_id);
client.last_sent.remove(&pty_id);
client.preview_next_send_at.remove(&pty_id);
client.scroll_offsets.remove(&pty_id);
client.scroll_caches.remove(&pty_id);
let removed_view = client.view_sizes.remove(&pty_id).is_some();
if client.lead == Some(pty_id) {
client.lead = None;
}
removed_sub || removed_view
}
fn update_client_scroll_state(client: &mut ClientState, pty_id: u16, next_offset: usize) -> bool {
let prev_offset = client.scroll_offsets.get(&pty_id).copied().unwrap_or(0);
if prev_offset == next_offset {
return false;
}
if prev_offset == 0 && next_offset > 0 {
client.scroll_caches.insert(
pty_id,
client.last_sent.get(&pty_id).cloned().unwrap_or_default(),
);
} else if prev_offset > 0
&& next_offset == 0
&& let Some(cache) = client.scroll_caches.remove(&pty_id)
{
if cache.rows() > 0 && cache.cols() > 0 {
client.last_sent.insert(pty_id, cache);
} else {
client.last_sent.remove(&pty_id);
}
}
if next_offset > 0 {
client.scroll_offsets.insert(pty_id, next_offset);
} else {
client.scroll_offsets.remove(&pty_id);
}
reset_inflight(client);
true
}
struct Session {
ptys: HashMap<u16, Pty>,
compositor: Option<SharedCompositor>,
next_client_id: u64,
next_compositor_id: u16,
next_pty_id: u16,
tick_fires: u32,
tick_snaps: u32,
surface_commits: u32,
surface_encodes: u32,
surface_encode_bytes: u64,
surface_frames_sent: u32,
/// Ticks where pixel_snapshot was empty → entire encode loop skipped.
ticks_pixel_snapshot_empty: u32,
/// Number of (sid,w,h) tuples in the most recent non-empty pixel_snapshot.
pixel_snapshot_len: usize,
last_ping: Instant,
clients: HashMap<u64, ClientState>,
}
struct SearchResultRow {
pty_id: u16,
score: u32,
primary_source: u8,
matched_sources: u8,
context: String,
scroll_offset: Option<usize>,
}
struct TickOutcome {
next_deadline: Option<Instant>,
}
impl Session {
fn new() -> Self {
Self {
ptys: HashMap::new(),
compositor: None,
next_client_id: 1,
next_compositor_id: 1,
next_pty_id: 1,
clients: HashMap::new(),
tick_fires: 0,
tick_snaps: 0,
surface_commits: 0,
surface_encodes: 0,
surface_encode_bytes: 0,
ticks_pixel_snapshot_empty: 0,
pixel_snapshot_len: 0,
last_ping: Instant::now(),
surface_frames_sent: 0,
}
}
fn ensure_compositor(
&mut self,
verbose: bool,
event_notify: Arc<dyn Fn() + Send + Sync>,
gpu_device: &str,
) -> &str {
if self.compositor.is_none() {
let session_id = self.next_compositor_id;
self.next_compositor_id = self.next_compositor_id.wrapping_add(1);
// Create the epoch before spawning anything so audio and video
// share the same time origin for A/V sync.
let created_at = Instant::now();
let handle = blit_compositor::spawn_compositor(verbose, event_notify, gpu_device);
#[cfg(target_os = "linux")]
let audio_broadcast = audio::AudioBroadcast::new();
#[cfg(target_os = "linux")]
let audio_pipeline = {
let audio_disabled = std::env::var("BLIT_AUDIO")
.map(|v| v == "0")
.unwrap_or(false);
if !audio_disabled && audio::pipewire_available() {
let runtime_dir = std::path::Path::new(&handle.socket_name)
.parent()
.unwrap_or(std::path::Path::new("/tmp"));
let bitrate = std::env::var("BLIT_AUDIO_BITRATE")
.ok()
.and_then(|v| v.parse::<i32>().ok())
.unwrap_or(0);
// Wrap in block_in_place so the thread::sleep calls
// inside spawn() don't stall the tokio runtime.
let broadcast = audio_broadcast.clone();
tokio::task::block_in_place(|| {
match audio::AudioPipeline::spawn(
runtime_dir,
session_id,
bitrate,
verbose,
created_at,
broadcast,
) {
Ok(pipeline) => {
if verbose {
eprintln!(
"[audio] pipeline started, PULSE_SERVER={}",
pipeline.pulse_server_path(),
);
}
Some(pipeline)
}
Err(e) => {
eprintln!("[audio] failed to start pipeline: {e}");
None
}
}
})
} else {
if verbose && !audio_disabled {
let missing = audio::missing_pipewire_binaries();
let load_err = audio_pw::load_error();
if !missing.is_empty() {
eprintln!(
"[audio] audio disabled: missing binaries on $PATH: {}",
missing.join(", ")
);
}
if !load_err.is_empty() {
eprintln!("[audio] audio disabled: {load_err}");
}
if missing.is_empty() && load_err.is_empty() {
eprintln!(
"[audio] audio disabled (reason not recorded; call pipewire_available() logged above)"
);
}
}
None
}
};
self.compositor = Some(SharedCompositor {
handle,
surfaces: HashMap::new(),
last_pixels: HashMap::new(),
last_frame_request: HashMap::new(),
created_at,
pixel_generation: 0,
last_blanket_frame_request: Instant::now(),
last_configured_size: HashMap::new(),
#[cfg(target_os = "linux")]
audio_pipeline,
#[cfg(target_os = "linux")]
audio_broadcast,
#[cfg(target_os = "linux")]
audio_session_id: session_id,
#[cfg(target_os = "linux")]
last_audio_restart: None,
});
}
&self.compositor.as_ref().unwrap().handle.socket_name
}
/// Returns the `PULSE_SERVER` path if the audio pipeline is active.
#[cfg(target_os = "linux")]
fn pulse_server_path(&self) -> Option<String> {
self.compositor
.as_ref()
.and_then(|cs| cs.audio_pipeline.as_ref())
.map(|ap| ap.pulse_server_path())
}
/// Returns the `PIPEWIRE_REMOTE` path if the audio pipeline is active.
#[cfg(target_os = "linux")]
fn pipewire_remote_path(&self) -> Option<String> {
self.compositor
.as_ref()
.and_then(|cs| cs.audio_pipeline.as_ref())
.map(|ap| ap.pipewire_remote_path())
}
fn allocate_pty_id(&mut self, max_ptys: usize) -> Option<u16> {
if max_ptys > 0 && self.ptys.len() >= max_ptys {
return None;
}
let start = self.next_pty_id;
let mut id = start;
loop {
if !self.ptys.contains_key(&id) {
self.next_pty_id = if id == u16::MAX { 1 } else { id + 1 };
return Some(id);
}
id = if id == u16::MAX { 1 } else { id + 1 };
if id == start {
return None;
}
}
}
fn send_to_all(&self, msg: &[u8]) {
for c in self.clients.values() {
let _ = send_outbox(c, msg.to_vec());
}
}
fn mediated_size_for_pty(&self, pty_id: u16) -> Option<(u16, u16)> {
let mut min_rows: Option<u16> = None;
let mut min_cols: Option<u16> = None;
for c in self.clients.values() {
if let Some((r, cols)) = c.view_sizes.get(&pty_id).copied() {
min_rows = Some(min_rows.map_or(r, |m: u16| m.min(r)));
min_cols = Some(min_cols.map_or(cols, |m: u16| m.min(cols)));
}
}
match (min_rows, min_cols) {
(Some(r), Some(c)) => Some((r.max(1), c.max(1))),
_ => None,
}
}
fn resize_pty(&mut self, pty_id: u16, rows: u16, cols: u16) -> bool {
let pty = match self.ptys.get_mut(&pty_id) {
Some(p) => p,
None => return false,
};
let (cur_rows, cur_cols) = pty.driver.size();
if cur_rows == rows && cur_cols == cols {
return false;
}
pty.ready_frames.clear();
pty.driver.resize(rows, cols);
pty.mark_dirty();
for c in self.clients.values_mut() {
if c.subscriptions.contains(&pty_id) {
c.last_sent.remove(&pty_id);
}
if c.scroll_caches.remove(&pty_id).is_some() {
reset_inflight(c);
}
}
if !pty.exited {
pty::resize_pty_os(&pty.handle, rows, cols);
}
true
}
fn resize_ptys_to_mediated_sizes<I>(&mut self, pty_ids: I) -> bool
where
I: IntoIterator<Item = u16>,
{
let mut changed = false;
let mut seen = HashSet::new();
for pty_id in pty_ids {
if !seen.insert(pty_id) {
continue;
}
if let Some((rows, cols)) = self.mediated_size_for_pty(pty_id) {
changed |= self.resize_pty(pty_id, rows, cols);
}
}
changed
}
// ------------------------------------------------------------------
// Surface sizing — same consumer-tracking model as PTY sizing.
// Each client reports how large it can display a surface; the server
// picks min(width), min(height) across all clients and configures the
// compositor accordingly.
// ------------------------------------------------------------------
/// Returns (width, height, scale_120) mediated across all clients.
/// Resolution: min across clients. DPI: max across clients.
/// Clients with a fixed per-surface target size (scaled subscription)
/// are excluded — they don't pull the compositor surface smaller for
/// anyone else; the server scales the native frame for them.
fn mediated_size_for_surface(
&self,
surface_id: u16,
max: Option<(u16, u16)>,
) -> Option<(u16, u16, u16)> {
let mut min_w: Option<u16> = None;
let mut min_h: Option<u16> = None;
let mut max_scale: u16 = 0;
for c in self.clients.values() {
if let Some(&(w, h, s)) = c.surface_view_sizes.get(&surface_id) {
min_w = Some(min_w.map_or(w, |m: u16| m.min(w)));
min_h = Some(min_h.map_or(h, |m: u16| m.min(h)));
max_scale = max_scale.max(s);
}
}
match (min_w, min_h) {
(Some(w), Some(h)) => {
let (w, h) = (w.max(1), h.max(1));
let (w, h) = if let Some((mw, mh)) = max {
(w.min(mw), h.min(mh))
} else {
(w, h)
};
Some((w, h, max_scale))
}
_ => None,
}
}
fn resize_surface(&mut self, surface_id: u16, width: u16, height: u16, scale_120: u16) -> bool {
let cs = match self.compositor.as_mut() {
Some(cs) => cs,
None => return false,
};
// Dedup against the last *requested* dimensions, not the composited
// output dimensions (`info.width`/`info.height`). The composited
// output may be smaller when the Wayland client sets xdg_geometry
// (e.g. Chromium excludes the title bar), so comparing against it
// would cause every resize to look like a change, flooding the
// compositor with redundant configures and re-creating the encoder
// (keyframe) on every tick during a drag-resize.
if let Some(&(lw, lh, ls)) = cs.last_configured_size.get(&surface_id)
&& lw == width
&& lh == height
&& ls == scale_120
{
return false;
}
cs.last_configured_size
.insert(surface_id, (width, height, scale_120));
let _ = cs.handle.command_tx.send(CompositorCommand::SurfaceResize {
surface_id,
width,
height,
scale_120,
});
true
}
fn resize_surfaces_to_mediated_sizes<I>(
&mut self,
surface_ids: I,
encoder_preferences: &[SurfaceEncoderPreference],
) where
I: IntoIterator<Item = u16>,
{
let max = SurfaceEncoderPreference::max_dimensions_for_list(encoder_preferences);
let mut seen = HashSet::new();
for sid in surface_ids {
if !seen.insert(sid) {
continue;
}
if let Some((w, h, scale_120)) = self.mediated_size_for_surface(sid, max) {
self.resize_surface(sid, w, h, scale_120);
}
}
}
fn pty_list_msg(&self) -> Vec<u8> {
let mut msg = vec![S2C_LIST];
let count = self.ptys.len() as u16;
msg.extend_from_slice(&count.to_le_bytes());
let mut ids: Vec<u16> = self.ptys.keys().copied().collect();
ids.sort();
for id in ids {
let pty = &self.ptys[&id];
let tag = pty.tag.as_bytes();
msg.extend_from_slice(&id.to_le_bytes());
msg.extend_from_slice(&(tag.len() as u16).to_le_bytes());
msg.extend_from_slice(tag);
let cmd = pty.command.as_deref().unwrap_or("").as_bytes();
msg.extend_from_slice(&(cmd.len() as u16).to_le_bytes());
msg.extend_from_slice(cmd);
}
msg
}
fn surface_list_msg(&self) -> Vec<u8> {
let cs = match self.compositor.as_ref() {
Some(cs) => cs,
None => {
let mut msg = vec![S2C_SURFACE_LIST];
msg.extend_from_slice(&0u16.to_le_bytes());
return msg;
}
};
let mut msg = vec![S2C_SURFACE_LIST];
let count = cs.surfaces.len() as u16;
msg.extend_from_slice(&count.to_le_bytes());
let mut ids: Vec<u16> = cs.surfaces.keys().copied().collect();
ids.sort();
for id in ids {
let info = &cs.surfaces[&id];
let title = info.title.as_bytes();
let app_id = info.app_id.as_bytes();
msg.extend_from_slice(&info.surface_id.to_le_bytes());
msg.extend_from_slice(&info.parent_id.to_le_bytes());
msg.extend_from_slice(&info.width.to_le_bytes());
msg.extend_from_slice(&info.height.to_le_bytes());
msg.extend_from_slice(&(title.len() as u16).to_le_bytes());
msg.extend_from_slice(title);
msg.extend_from_slice(&(app_id.len() as u16).to_le_bytes());
msg.extend_from_slice(app_id);
}
msg
}
}
struct AppStateInner {
config: Config,
session: Mutex<Session>,
pty_fds: PtyFds,
delivery_notify: Arc<Notify>,
/// Signalled when a client sends C2S_QUIT to initiate server shutdown.
shutdown_notify: Arc<Notify>,
/// Tracks the number of currently connected clients for enforcing
/// `config.max_connections`.
active_connections: std::sync::atomic::AtomicUsize,
}
type AppState = Arc<AppStateInner>;
fn nudge_delivery(state: &AppState) {
state.delivery_notify.notify_one();
}
#[cfg(unix)]
#[allow(dead_code)]
fn spawn_compositor_child(
command: &str,
argv: Option<&[&str]>,
wayland_socket: &str,
dir: Option<&str>,
) -> libc::pid_t {
use std::ffi::CString;
let pid = unsafe { libc::fork() };
if pid == 0 {
if let Some(d) = dir {
let c_dir = CString::new(d).unwrap();
unsafe {
libc::chdir(c_dir.as_ptr());
}
}
unsafe {
let wd_path = std::path::Path::new(wayland_socket);
if let Some(dir) = wd_path.parent() {
let xdg = std::env::var_os("XDG_RUNTIME_DIR");
let needs_update = match &xdg {
Some(x) => std::path::Path::new(x) != dir,
None => true,
};
if needs_update {
std::env::set_var("XDG_RUNTIME_DIR", dir);
}
}
std::env::set_var("WAYLAND_DISPLAY", wayland_socket);
std::env::remove_var("DISPLAY");
std::env::remove_var("DBUS_SESSION_BUS_ADDRESS");
std::env::remove_var("DBUS_SYSTEM_BUS_ADDRESS");
}
if let Some(args) = argv {
let prog = CString::new(args[0]).unwrap();
let c_args: Vec<CString> = args.iter().map(|a| CString::new(*a).unwrap()).collect();
let c_ptrs: Vec<*const libc::c_char> = c_args
.iter()
.map(|a| a.as_ptr())
.chain(std::iter::once(std::ptr::null()))
.collect();
unsafe {
libc::execvp(prog.as_ptr(), c_ptrs.as_ptr());
}
} else {
let prog = CString::new(command).unwrap();
let c_ptrs = [prog.as_ptr(), std::ptr::null()];
unsafe {
libc::execvp(prog.as_ptr(), c_ptrs.as_ptr());
libc::_exit(1);
}
}
}
pid
}
/// Map xterm-256 color index to (r, g, b) in 16-bit per channel.
fn xterm256_color(idx: u8) -> (u16, u16, u16) {
// Standard 16 colors (0-15)
const BASE16: [(u8, u8, u8); 16] = [
(0, 0, 0),
(128, 0, 0),
(0, 128, 0),
(128, 128, 0),
(0, 0, 128),
(128, 0, 128),
(0, 128, 128),
(192, 192, 192),
(128, 128, 128),
(255, 0, 0),
(0, 255, 0),
(255, 255, 0),
(0, 0, 255),
(255, 0, 255),
(0, 255, 255),
(255, 255, 255),
];
let (r8, g8, b8) = if idx < 16 {
BASE16[idx as usize]
} else if idx < 232 {
// 6x6x6 color cube (indices 16-231)
let n = idx - 16;
let ri = n / 36;
let gi = (n % 36) / 6;
let bi = n % 6;
let to_val = |v: u8| if v == 0 { 0u8 } else { 55 + 40 * v };
(to_val(ri), to_val(gi), to_val(bi))
} else {
// Grayscale ramp (indices 232-255)
let v = 8 + 10 * (idx - 232);
(v, v, v)
};
// Scale 8-bit to 16-bit (0xFF -> 0xFFFF)
let scale = |v: u8| (v as u16) << 8 | v as u16;
(scale(r8), scale(g8), scale(b8))
}
fn parse_terminal_queries(data: &[u8], size: (u16, u16), cursor: (u16, u16)) -> Vec<String> {
const DA1_RESPONSE: &[u8] = b"\x1b[?64;1;2;6;9;15;18;21;22c";
let mut results = Vec::new();
let mut i = 0;
while i < data.len() {
if data[i] != 0x1b || i + 1 >= data.len() {
i += 1;
continue;
}
// Handle OSC sequences: \x1b] ... (ST or BEL)
if data[i + 1] == b']' {
let osc_start = i + 2;
// Find the terminator: BEL (\x07) or ST (\x1b\\)
let mut end = osc_start;
while end < data.len() {
if data[end] == 0x07 {
break;
}
if data[end] == 0x1b && end + 1 < data.len() && data[end + 1] == b'\\' {
break;
}
end += 1;
}
if end < data.len() {
let payload = &data[osc_start..end];
// OSC 11 ; ? — query background color
if payload == b"11;?" {
// Respond with dark background (rgb:0000/0000/0000)
results.push("\x1b]11;rgb:0000/0000/0000\x1b\\".into());
}
// OSC 10 ; ? — query foreground color
else if payload == b"10;?" {
results.push("\x1b]10;rgb:ffff/ffff/ffff\x1b\\".into());
}
// OSC 4 ; N ; ? — query palette color N
else if payload.starts_with(b"4;") && payload.ends_with(b";?") {
let idx_bytes = &payload[2..payload.len() - 2];
if let Ok(idx_str) = std::str::from_utf8(idx_bytes)
&& let Ok(idx) = idx_str.parse::<u8>()
{
let (r, g, b) = xterm256_color(idx);
results.push(format!("\x1b]4;{idx};rgb:{r:04x}/{g:04x}/{b:04x}\x1b\\"));
}
}
i = end + if data[end] == 0x07 { 1 } else { 2 };
continue;
}
i = end;
continue;
}
// Handle CSI sequences: \x1b[ ...
if i + 2 >= data.len() || data[i + 1] != b'[' {
i += 1;
continue;
}
i += 2;
let has_q = i < data.len() && data[i] == b'?';
if has_q {
i += 1;
}
let param_start = i;
while i < data.len() && (data[i].is_ascii_digit() || data[i] == b';') {
i += 1;
}
if i >= data.len() {
break;
}
let final_byte = data[i];
let params = &data[param_start..i];
i += 1;
if has_q {
continue;
}
let resp: Option<String> = match final_byte {
b'c' if params.is_empty() || params == b"0" => {
Some(String::from_utf8_lossy(DA1_RESPONSE).into_owned())
}
b'n' if params == b"6" => Some(format!("\x1b[{};{}R", cursor.0 + 1, cursor.1 + 1)),
b'n' if params == b"5" => Some("\x1b[0n".into()),
b't' if params == b"18" => {
let (rows, cols) = size;
Some(format!("\x1b[8;{rows};{cols}t"))
}
b't' if params == b"14" => {
let (rows, cols) = size;
Some(format!("\x1b[4;{};{}t", rows * 16, cols * 8))
}
_ => None,
};
if let Some(r) = resp {
results.push(r);
}
}
results
}
async fn cleanup_pty_internal(pty_id: u16, state: &AppState) {
state.pty_fds.write().unwrap().remove(&pty_id);
let mut sess = state.session.lock().await;
if let Some(pty) = sess.ptys.get_mut(&pty_id) {
if pty.exited {
return;
}
pty.exited = true;
pty::close_pty(&pty.handle);
pty.exit_status = pty::collect_exit_status(&pty.handle);
pty.mark_dirty();
let msg = blit_remote::msg_exited(pty_id, pty.exit_status);
sess.send_to_all(&msg);
}
}
fn take_snapshot(pty: &mut Pty) -> FrameState {
if pty.lflag_last.elapsed() >= Duration::from_millis(250) {
pty.lflag_cache = pty::pty_lflag(&pty.handle);
pty.lflag_last = Instant::now();
}
let (echo, icanon) = pty.lflag_cache;
pty.driver.snapshot(echo, icanon)
}
fn build_scrollback_update(
pty: &mut Pty,
id: u16,
offset: usize,
prev_frame: &FrameState,
) -> Option<(Vec<u8>, FrameState)> {
let frame = pty.driver.scrollback_frame(offset);
let msg = build_update_msg(id, &frame, prev_frame);
msg.map(|m| (m, frame))
}
fn build_search_results_msg(request_id: u16, results: &[SearchResultRow]) -> Vec<u8> {
let count = results.len().min(u16::MAX as usize);
let payload_bytes: usize = results[..count]
.iter()
.map(|result| 14 + result.context.len().min(u16::MAX as usize))
.sum();
let mut msg = Vec::with_capacity(5 + payload_bytes);
msg.push(S2C_SEARCH_RESULTS);
msg.extend_from_slice(&request_id.to_le_bytes());
msg.extend_from_slice(&(count as u16).to_le_bytes());
for result in &results[..count] {
msg.extend_from_slice(&result.pty_id.to_le_bytes());
msg.extend_from_slice(&result.score.to_le_bytes());
msg.push(result.primary_source);
msg.push(result.matched_sources);
let scroll_offset = result
.scroll_offset
.map(|offset| offset.min(u32::MAX as usize - 1) as u32)
.unwrap_or(u32::MAX);
msg.extend_from_slice(&scroll_offset.to_le_bytes());
let context = result.context.as_bytes();
let context_len = context.len().min(u16::MAX as usize);
msg.extend_from_slice(&(context_len as u16).to_le_bytes());
msg.extend_from_slice(&context[..context_len]);
}
msg
}
enum SendOutcome {
NoChange,
Sent,
Backpressured,
}
fn try_send_update(
client: &mut ClientState,
pid: u16,
current: FrameState,
msg: Option<Vec<u8>>,
now: Instant,
paced: bool,
) -> SendOutcome {
let Some(msg) = msg else {
return SendOutcome::NoChange;
};
let bytes = msg.len();
if send_outbox(client, msg).is_ok() {
client.last_sent.insert(pid, current);
record_send(client, bytes, now, paced);
client.frames_sent = client.frames_sent.wrapping_add(1);
SendOutcome::Sent
} else {
// Receiver dropped — client disconnected. Advance last_sent so
// the next diff (if any) is small rather than accumulating stale
// changes.
client.last_sent.insert(pid, current);
SendOutcome::Backpressured
}
}
pub async fn run(config: Config) {
let state: AppState = Arc::new(AppStateInner {
config,
session: Mutex::new(Session::new()),
pty_fds: Arc::new(std::sync::RwLock::new(HashMap::new())),
delivery_notify: Arc::new(Notify::new()),
shutdown_notify: Arc::new(Notify::new()),
active_connections: std::sync::atomic::AtomicUsize::new(0),
});
// Start the compositor eagerly so it is ready before any client
// connects or any terminal is created.
if !state.config.skip_compositor {
let notify = state.delivery_notify.clone();
let event_notify = Arc::new(move || notify.notify_one()) as Arc<dyn Fn() + Send + Sync>;
let mut sess = state.session.lock().await;
sess.ensure_compositor(
state.config.verbose,
event_notify,
&state.config.vaapi_device,
);
}
let delivery_state = state.clone();
tokio::spawn(async move {
let mut next_deadline: Option<Instant> = None;
loop {
if let Some(deadline) = next_deadline {
tokio::select! {
_ = delivery_state.delivery_notify.notified() => {}
_ = tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)) => {}
}
} else {
delivery_state.delivery_notify.notified().await;
}
let outcome = tick(&delivery_state).await;
next_deadline = outcome.next_deadline;
}
});
tokio::spawn(async {
loop {
tokio::time::sleep(Duration::from_secs(5)).await;
pty::reap_zombies();
}
});
#[cfg(unix)]
if let Some(channel_fd) = state.config.fd_channel {
ipc::run_fd_channel(channel_fd, state).await;
return;
}
#[cfg(unix)]
let listener = {
if let Some(l) = IpcListener::from_systemd_fd(state.config.verbose) {
l
} else {
IpcListener::bind(&state.config.ipc_path, state.config.verbose)
}
};
#[cfg(not(unix))]
let mut listener = IpcListener::bind(&state.config.ipc_path, state.config.verbose);
// Broadcast S2C_QUIT on SIGTERM / SIGINT so clients can reconnect promptly
// instead of waiting for a transport-level timeout.
{
let state = state.clone();
tokio::spawn(async move {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
let mut sigterm = signal(SignalKind::terminate()).expect("signal handler");
let mut sigint = signal(SignalKind::interrupt()).expect("signal handler");
tokio::select! {
_ = sigterm.recv() => {}
_ = sigint.recv() => {}
}
}
#[cfg(not(unix))]
{
let _ = tokio::signal::ctrl_c().await;
}
let sess = state.session.lock().await;
sess.send_to_all(&[S2C_QUIT]);
drop(sess);
state.shutdown_notify.notify_one();
});
}
let shutdown = state.shutdown_notify.clone();
loop {
let stream = tokio::select! {
result = listener.accept() => match result {
Ok(s) => s,
Err(e) => {
eprintln!("accept error: {e}");
tokio::time::sleep(Duration::from_millis(100)).await;
continue;
}
},
_ = shutdown.notified() => break,
};
let max = state.config.max_connections;
if max > 0 {
let current = state
.active_connections
.load(std::sync::atomic::Ordering::Relaxed);
if current >= max {
eprintln!("max connections ({max}) reached, rejecting");
drop(stream);
continue;
}
}
state
.active_connections
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let state = state.clone();
tokio::spawn(async move {
handle_client(stream, state.clone()).await;
state
.active_connections
.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
});
}
// Brief grace period for S2C_QUIT to reach clients before the process exits.
tokio::time::sleep(Duration::from_millis(100)).await;
}
/// Minimum interval between blanket RequestFrame rounds. Keeps video
/// players (mpv) and browsers ticking even when no client is consuming
/// frames. Also used as the maximum tick-loop sleep so the loop never
/// blocks longer than this.
///
/// When any client has an active surface subscription, use 8 ms (~120 Hz)
/// so video players get consistent frame callbacks matching the display
/// rate. Without active surfaces, 33 ms (30 Hz) is sufficient.
const BLANKET_FRAME_INTERVAL_IDLE: Duration = Duration::from_millis(250);
const BLANKET_FRAME_INTERVAL_SURFACE: Duration = Duration::from_micros(62_500);
/// Returns the interval at which the tick loop must send blanket
/// `RequestFrame` events to keep Wayland apps (mpv, browsers, etc.)
/// making progress. Returns `None` when no clients are connected — in
/// that state the loop can sleep purely on event notifications, and
/// apps pause until a viewer reconnects (resuming within SURFACE).
fn blanket_frame_interval(sess: &Session) -> Option<Duration> {
if sess.clients.is_empty() {
return None;
}
let has_surface_subs = sess
.clients
.values()
.any(|c| !c.surface_subscriptions.is_empty());
if has_surface_subs {
Some(BLANKET_FRAME_INTERVAL_SURFACE)
} else {
Some(BLANKET_FRAME_INTERVAL_IDLE)
}
}
async fn tick(state: &AppState) -> TickOutcome {
let mut sess = state.session.lock().await;
sess.tick_fires += 1;
let mut next_deadline: Option<Instant> = None;
let now = Instant::now();
// Emit pacing metrics every 10s for each client, even when no ACKs
// are flowing (idle session): the ACK handler also calls this so the
// first client with traffic still owns the tick-counter reset.
let log_client_ids: Vec<u64> = sess.clients.keys().copied().collect();
for cid in log_client_ids {
maybe_log_pacing_metrics(&mut sess, cid, state.config.verbose);
}
// Application-level keepalive. Only scheduled when a client is
// connected — otherwise there's no one to ping and the timer would
// be pure polling cost.
let ping_interval = state.config.ping_interval;
if !ping_interval.is_zero() && !sess.clients.is_empty() {
if now.duration_since(sess.last_ping) >= ping_interval {
sess.send_to_all(&[S2C_PING]);
sess.last_ping = now;
}
let next_ping = sess.last_ping + ping_interval;
next_deadline = Some(next_deadline.map_or(next_ping, |d: Instant| d.min(next_ping)));
}
// Surface IDs whose per-client encoders need to be invalidated.
let mut invalidate_client_encoders: Vec<u16> = Vec::new();
// Surface IDs resized by the compositor this tick. After the
// compositor borrow is released we wake pacing for every client
// subscribed to each sid so the first post-resize frame bypasses
// the per-surface time gate.
let mut resized_surface_ids: Vec<u16> = Vec::new();
let mut surface_commit_count = 0u32;
if let Some(cs) = sess.compositor.as_mut() {
let mut events = Vec::new();
while let Ok(event) = cs.handle.event_rx.try_recv() {
events.push(event);
}
let mut broadcast: Vec<Vec<u8>> = Vec::new();
for event in events {
match event {
CompositorEvent::SurfaceCreated {
surface_id,
title,
app_id,
parent_id,
width,
height,
} => {
broadcast.push(msg_surface_created(
surface_id, parent_id, width, height, &title, &app_id,
));
cs.surfaces.insert(
surface_id,
CachedSurfaceInfo {
surface_id,
parent_id,
width,
height,
title,
app_id,
},
);
cs.last_pixels.remove(&surface_id);
invalidate_client_encoders.push(surface_id);
}
CompositorEvent::SurfaceDestroyed { surface_id } => {
cs.surfaces.remove(&surface_id);
cs.last_pixels.remove(&surface_id);
cs.last_configured_size.remove(&surface_id);
invalidate_client_encoders.push(surface_id);
broadcast.push(msg_surface_destroyed(surface_id));
}
CompositorEvent::SurfaceCommit {
surface_id,
width,
height,
pixels,
timestamp_ms,
} => {
surface_commit_count += 1;
if let Some(info) = cs.surfaces.get_mut(&surface_id) {
info.width = width as u16;
info.height = height as u16;
}
cs.pixel_generation += 1;
cs.last_pixels.insert(
surface_id,
LastPixels {
width,
height,
pixels,
generation: cs.pixel_generation,
timestamp_ms,
},
);
}
CompositorEvent::SurfaceTitle { surface_id, title } => {
if let Some(info) = cs.surfaces.get_mut(&surface_id) {
info.title = title.clone();
}
broadcast.push(msg_surface_title(surface_id, &title));
}
CompositorEvent::SurfaceAppId { surface_id, app_id } => {
if let Some(info) = cs.surfaces.get_mut(&surface_id) {
info.app_id = app_id.clone();
}
broadcast.push(msg_surface_app_id(surface_id, &app_id));
}
CompositorEvent::SurfaceResized {
surface_id,
width,
height,
} => {
if let Some(info) = cs.surfaces.get_mut(&surface_id) {
info.width = width;
info.height = height;
}
cs.last_pixels.remove(&surface_id);
// Don't eagerly invalidate client encoders here. The
// encode path already checks for dimension mismatches
// (source_dimensions != pixel size) and recreates the
// encoder on demand. Eagerly destroying encoders on
// every intermediate size during a drag-resize causes
// expensive encoder teardown+creation cycles for sizes
// that may never actually be encoded (because a newer
// SurfaceCommit arrives before the next encode tick).
broadcast.push(msg_surface_resized(surface_id, width, height));
resized_surface_ids.push(surface_id);
}
CompositorEvent::ClipboardContent {
mime_type, data, ..
} => {
broadcast.push(msg_s2c_clipboard_content(&mime_type, &data));
}
CompositorEvent::SurfaceCursor { surface_id, cursor } => {
// Format: [0x29][surface_id:2][type:1][payload...]
// type 0 = named: [name_len:1][name:N]
// type 1 = hidden (no payload)
// type 2 = custom: [hotx:2][hoty:2][w:2][h:2][png:N]
let mut msg = Vec::new();
msg.push(blit_remote::S2C_SURFACE_CURSOR);
msg.extend_from_slice(&surface_id.to_le_bytes());
match &cursor {
blit_compositor::CursorImage::Named(name) => {
msg.push(0); // type = named
msg.push(name.len() as u8);
msg.extend_from_slice(name.as_bytes());
}
blit_compositor::CursorImage::Hidden => {
msg.push(1); // type = hidden
}
blit_compositor::CursorImage::Custom {
hotspot_x,
hotspot_y,
width,
height,
rgba,
} => {
// Encode as PNG to keep message small.
let mut png_buf = Vec::new();
{
let mut encoder =
png::Encoder::new(&mut png_buf, *width as u32, *height as u32);
encoder.set_color(png::ColorType::Rgba);
encoder.set_depth(png::BitDepth::Eight);
if let Ok(mut writer) = encoder.write_header() {
let _ = writer.write_image_data(rgba);
}
}
msg.push(2); // type = custom
msg.extend_from_slice(&hotspot_x.to_le_bytes());
msg.extend_from_slice(&hotspot_y.to_le_bytes());
msg.extend_from_slice(&width.to_le_bytes());
msg.extend_from_slice(&height.to_le_bytes());
msg.extend_from_slice(&png_buf);
}
}
broadcast.push(msg);
}
}
}
for msg in &broadcast {
sess.send_to_all(msg);
}
}
sess.surface_commits += surface_commit_count;
// Apply deferred per-client encoder invalidation (couldn't mutate
// sess.clients while sess.compositor was borrowed above). Any
// surface event (resize, destroy, reconfigure) invalidates every
// encoder bound to that sid's pixel stream.
for sid in invalidate_client_encoders {
for c in sess.clients.values_mut() {
c.surface_subs.remove(&sid);
c.vulkan_video_surfaces.remove(&sid);
}
}
// Wake pacing for every subscriber of a compositor-resized surface.
// Reset the burst window and clear next_send_at so the first frame
// at the new dimensions flows at wire speed instead of waiting for
// the per-surface time gate (up to ~1/fps), and force a keyframe
// so decoders recover cleanly after the dimension change.
for sid in resized_surface_ids {
for c in sess.clients.values_mut() {
if !c.surface_subscriptions.contains(&sid) {
continue;
}
if let Some(s) = c.surface_subs.get_mut(&sid) {
s.burst_remaining = SURFACE_BURST_FRAMES;
s.next_send_at = None;
s.nal_none_streak = 0;
s.nal_none_latched_at = None;
}
c.surface_needs_keyframe = true;
}
}
// Per-client surface encode + deliver.
// Each client has its own encoder per surface. We encode from
// shared last_pixels into each client's encoder and deliver.
//
// Snapshot pixel metadata from the compositor first to avoid
// holding an immutable borrow on sess.compositor while mutating
// sess.clients.
// Snapshot every surface entry so each client's per-surface encoder
// can draw from the latest pixels without holding the compositor
// borrow through the (lengthy) encoder-dispatch loop below.
let pixel_snapshot: Vec<(u16, u32, u32, u64, u32)> = sess
.compositor
.as_ref()
.map(|cs| {
cs.last_pixels
.iter()
.map(|(&sid, lp)| (sid, lp.width, lp.height, lp.generation, lp.timestamp_ms))
.collect()
})
.unwrap_or_default();
if pixel_snapshot.is_empty() {
sess.ticks_pixel_snapshot_empty = sess.ticks_pixel_snapshot_empty.saturating_add(1);
} else {
sess.pixel_snapshot_len = pixel_snapshot.len();
}
// ---- Surface encode (off main thread) + deliver ----
//
// Collect encode jobs, drop the session lock, run encodes in
// spawn_blocking, re-acquire the lock, and deliver.
struct EncodeJob {
cid: u64,
sid: u16,
/// Dimensions the encoder operates at — always the compositor's
/// native pixel size.
px_w: u32,
px_h: u32,
/// Pixel data to encode.
pixels: blit_compositor::PixelData,
needs_keyframe: bool,
encoder: SurfaceEncoder,
generation: u64,
/// CLOCK_MONOTONIC ms captured at compositor commit time.
timestamp_ms: u32,
}
struct EncoderCreateParams {
preferences: Vec<SurfaceEncoderPreference>,
vaapi_device: String,
quality: SurfaceQuality,
verbose: bool,
codec_support: u8,
chroma: ChromaSubsampling,
}
/// A creation task runs `SurfaceEncoder::new` + GBM-buffer
/// allocation on a blocking thread, then hands back the encoder
/// and its external buffers to the main loop to register with the
/// compositor. No encoding happens here — the first encode runs
/// on a subsequent tick after the compositor has committed into
/// the new buffers.
struct CreateJob {
cid: u64,
sid: u16,
px_w: u32,
px_h: u32,
params: EncoderCreateParams,
}
struct CreateResult {
cid: u64,
sid: u16,
/// None when `SurfaceEncoder::new` failed; the completion
/// handler logs and latches a backoff so the tick loop doesn't
/// spin on retries.
encoder: Option<SurfaceEncoder>,
fresh: Option<FreshEncoder>,
}
/// Metadata shipped with an encode result when the encoder was
/// created this tick (deferred to spawn_blocking). `Some` = the
/// main loop should send S2C_SURFACE_ENCODER, register external
/// GBM buffers with the compositor, and accept the encoder back.
struct FreshEncoder {
name: &'static str,
codec_string: String,
#[cfg(target_os = "linux")]
external_bufs: Vec<blit_compositor::ExternalOutputBuffer>,
}
struct EncodeResult {
cid: u64,
sid: u16,
/// Encoded frame dimensions (what goes on the wire).
px_w: u32,
px_h: u32,
generation: u64,
encoder: SurfaceEncoder,
nal_data: Option<(Vec<u8>, bool)>, // (data, is_keyframe)
codec_flag: u8,
/// CLOCK_MONOTONIC ms from compositor commit time.
timestamp_ms: u32,
}
let mut encode_jobs: Vec<EncodeJob> = Vec::new();
let mut create_jobs: Vec<CreateJob> = Vec::new();
// Surfaces that had encode jobs dispatched this tick. Used below to
// eagerly pre-request the next frame so the compositor renders in
// parallel with the in-flight encode (pipeline overlap).
let mut encode_dispatched_surfaces: HashSet<u16> = HashSet::new();
// Collect (cid, subs, needs_kf) for clients that are due, then build
// encode jobs in a second pass to avoid overlapping borrows. `subs`
// is the set of surface ids this client subscribes to.
struct ClientWork {
cid: u64,
subs: HashSet<u16>,
needs_keyframe: bool,
}
let mut client_work: Vec<ClientWork> = Vec::new();
if !pixel_snapshot.is_empty() {
for (&cid, client) in sess.clients.iter_mut() {
if !surface_window_open(client) {
// Log persistent blockage so hangs are visible.
let now_inst = Instant::now();
if now_inst
.duration_since(client.last_window_blocked_log)
.as_secs_f32()
> 5.0
{
client.last_window_blocked_log = now_inst;
let max_burst: u8 = client
.surface_subs
.values()
.map(|s| s.burst_remaining)
.max()
.unwrap_or(0);
eprintln!(
"[surface-gate] cid={cid} surface_window_open=false outbox={}f/{}B (limits {}f/{}B) burst={max_burst}",
outbox_queued_frames(client),
outbox_queued_bytes(client),
OUTBOX_SOFT_QUEUE_LIMIT_FRAMES,
OUTBOX_SOFT_QUEUE_LIMIT_BYTES,
);
}
continue;
}
// Per-surface pacing is checked in the inner loop below so
// that each surface can run at full frame rate independently.
if client.surface_subscriptions.is_empty() {
client.skip_no_subs_count = client.skip_no_subs_count.saturating_add(1);
continue;
}
let subs: HashSet<u16> = client.surface_subscriptions.iter().copied().collect();
client_work.push(ClientWork {
cid,
subs,
needs_keyframe: client.surface_needs_keyframe,
});
// Don't advance the deadline here — wait until we know an
// encode job was actually collected (see below). Advancing
// eagerly wastes time slots when the encode is skipped due
// to in-flight limits or unchanged pixel data.
}
// Track which (client, surface) pairs actually had encode jobs
// collected so we can advance per-surface deadlines afterwards.
let mut encoded_client_surfaces: HashSet<(u64, u16)> = HashSet::new();
// Pre-extract compositor Vulkan Video capabilities so we don't
// need to borrow sess.compositor inside the client-mutation loop.
let vk_encode_available = sess
.compositor
.as_ref()
.is_some_and(|cs| cs.handle.vulkan_video_encode);
let vk_encode_av1_available = sess
.compositor
.as_ref()
.is_some_and(|cs| cs.handle.vulkan_video_encode_av1);
// Vulkan Video encoder setup commands to send after the client loop.
struct VulkanEncoderSetup {
surface_id: u32,
codec: u8,
qp: u8,
width: u32,
height: u32,
}
let mut pending_vulkan_encoder_setups: Vec<VulkanEncoderSetup> = Vec::new();
let mut pending_vulkan_keyframe_requests: Vec<u32> = Vec::new();
for work in &client_work {
for &sid in &work.subs {
let Some(&(_, px_w, px_h, px_gen, px_timestamp_ms)) =
pixel_snapshot.iter().find(|&&(s, _, _, _, _)| s == sid)
else {
let client = sess.clients.get_mut(&work.cid).unwrap();
client.skip_last_pixels_mismatch_count =
client.skip_last_pixels_mismatch_count.saturating_add(1);
continue;
};
{
let client = sess.clients.get_mut(&work.cid).unwrap();
client.encode_loop_iters = client.encode_loop_iters.saturating_add(1);
}
let client = sess.clients.get_mut(&work.cid).unwrap();
// Per-surface pacing gate: during burst-start, skip the
// time-based check so frames flow at wire speed; otherwise
// each surface independently waits for its own deadline.
{
let (burst, deadline) = client.surface_subs.get(&sid).map_or((0, now), |s| {
(s.burst_remaining, s.next_send_at.unwrap_or(now))
});
if burst == 0 && deadline > now {
// Safety clamp: the deadline should never be more
// than 2× the send interval ahead. If it is, snap
// back to now so encoding doesn't stall permanently.
let interval = surface_send_interval(client);
if deadline > now + interval + interval {
client.surface_subs.entry(sid).or_default().next_send_at = Some(now);
} else {
next_deadline = Some(match next_deadline {
Some(existing) => existing.min(deadline),
None => deadline,
});
client.skip_pacing_count = client.skip_pacing_count.saturating_add(1);
continue;
}
}
}
// Skip encoding if the pixel data hasn't changed since the
// last encode for this surface, unless a keyframe is needed.
if !work.needs_keyframe
&& let Some(last_gen) = client
.surface_subs
.get(&sid)
.and_then(|s| s.last_encoded_gen)
&& last_gen == px_gen
{
client.skip_same_gen_count = client.skip_same_gen_count.saturating_add(1);
continue;
}
let pixels: blit_compositor::PixelData = {
let cs = sess.compositor.as_ref().unwrap();
match cs.last_pixels.get(&sid) {
Some(lp) if lp.width == px_w && lp.height == px_h => lp.pixels.clone(),
_ => {
let client = sess.clients.get_mut(&work.cid).unwrap();
client.skip_last_pixels_mismatch_count =
client.skip_last_pixels_mismatch_count.saturating_add(1);
continue;
}
}
};
let client = sess.clients.get_mut(&work.cid).unwrap();
let (enc_w, enc_h) = (px_w, px_h);
// Fast path: if the compositor already produced an encoded
// bitstream (Vulkan Video), skip the SurfaceEncoder entirely
// and send the pre-encoded data directly to the client.
if let blit_compositor::PixelData::Encoded {
ref data,
is_keyframe,
codec_flag,
} = pixels
{
let flags = codec_flag
| if is_keyframe {
SURFACE_FRAME_FLAG_KEYFRAME
} else {
0
};
let msg = msg_surface_frame(
sid,
px_timestamp_ms,
flags,
px_w as u16,
px_h as u16,
data,
);
let bytes = msg.len();
match send_outbox(client, msg) {
Err(_e) => {
client.surface_needs_keyframe = true;
}
Ok(()) => {
client.surface_inflight_frames.push_back(InFlightFrame {
sent_at: now,
bytes,
paced: true,
});
if !is_keyframe {
client.avg_surface_frame_bytes = ewma_with_direction(
client.avg_surface_frame_bytes,
bytes as f32,
0.5,
0.125,
);
}
client.frames_sent = client.frames_sent.wrapping_add(1);
if client.surface_needs_keyframe && is_keyframe {
client.surface_needs_keyframe = false;
}
if let Some(s) = client.surface_subs.get_mut(&sid) {
s.burst_remaining = s.burst_remaining.saturating_sub(1);
}
}
}
encoded_client_surfaces.insert((work.cid, sid));
encode_dispatched_surfaces.insert(sid);
client.surface_subs.entry(sid).or_default().last_encoded_gen = Some(px_gen);
continue;
}
// Skip if an encode or creation job is already in
// flight for this surface. Creations also block encode
// dispatch: the encoder is None while creation runs,
// and we don't want to re-queue another creation until
// the first one completes.
if client
.surface_subs
.get(&sid)
.is_some_and(|s| s.encode_in_flight || s.creation_in_flight)
{
client.skip_in_flight_count = client.skip_in_flight_count.saturating_add(1);
let now_inst = Instant::now();
if now_inst.duration_since(client.last_skip_log).as_secs_f32() > 5.0 {
client.last_skip_log = now_inst;
let burst = client
.surface_subs
.get(&sid)
.map_or(0, |s| s.burst_remaining);
eprintln!(
"[encode-skip] cid={} sid={sid} reason=in_flight same_gen={} in_flight={} burst={burst}",
work.cid, client.skip_same_gen_count, client.skip_in_flight_count,
);
}
continue;
}
let has_vulkan_enc = client.vulkan_video_surfaces.contains_key(&sid);
let needs_new_encoder = if has_vulkan_enc {
false
} else {
client
.surface_subs
.get(&sid)
.and_then(|s| s.encoder.as_ref())
.is_none_or(|e| e.source_dimensions() != (enc_w, enc_h))
};
// If the encoder was dropped due to persistent nal_data=None,
// back off for a short window before retrying. Each retry
// allocates GBM fds, so we don't want a genuinely broken
// encoder (GPU lost) to recreate at tick rate and exhaust
// the process fd limit — but a warm-up burst (compositor
// hasn't imported the freshly-allocated external output
// buffers yet) should recover within seconds without
// requiring a user-driven resize/resubscribe.
const NAL_NONE_RETRY_BACKOFF: Duration = Duration::from_secs(2);
if needs_new_encoder
&& client
.surface_subs
.get(&sid)
.is_some_and(|s| s.nal_none_streak >= 10)
{
let ready_to_retry = client
.surface_subs
.get(&sid)
.and_then(|s| s.nal_none_latched_at)
.is_some_and(|t| now.duration_since(t) >= NAL_NONE_RETRY_BACKOFF);
if ready_to_retry {
if let Some(s) = client.surface_subs.get_mut(&sid) {
s.nal_none_streak = 0;
s.nal_none_latched_at = None;
}
} else {
continue;
}
}
// --- Try Vulkan Video first ---
if needs_new_encoder {
let codec_support = client
.surface_subs
.get(&sid)
.map(|s| s.codec_override)
.filter(|&c| c != 0)
.unwrap_or(client.surface_codec_support);
let quality = client
.surface_subs
.get(&sid)
.and_then(|s| s.quality_override)
.unwrap_or(state.config.surface_quality);
for &pref in &state.config.surface_encoders {
if !pref.is_vulkan_video() {
continue;
}
if !pref.supported_by_client(codec_support) {
continue;
}
// Check compositor capability (pre-extracted above).
let available = match pref {
SurfaceEncoderPreference::VulkanVideoH264 => vk_encode_available,
SurfaceEncoderPreference::VulkanVideoAV1 => vk_encode_av1_available,
_ => false,
};
if !available {
continue;
}
// Vulkan Video 4:4:4 requires a YUV444 compute pipeline
// and 3-plane DPB surfaces that are not yet implemented.
// Skip so the fallback chain tries NVENC/VA-API/software.
if state.config.chroma.is_444() {
continue;
}
let qp = match pref {
SurfaceEncoderPreference::VulkanVideoAV1 => quality.av1_qp_for_vulkan(),
_ => quality.h264_qp(),
};
let enc_name: &'static str = match (pref, state.config.chroma) {
(
SurfaceEncoderPreference::VulkanVideoH264,
ChromaSubsampling::Cs444,
) => "h264-vulkan 4:4:4",
(SurfaceEncoderPreference::VulkanVideoH264, _) => "h264-vulkan",
(
SurfaceEncoderPreference::VulkanVideoAV1,
ChromaSubsampling::Cs444,
) => "av1-vulkan 4:4:4",
(SurfaceEncoderPreference::VulkanVideoAV1, _) => "av1-vulkan",
(_, ChromaSubsampling::Cs444) => "vulkan 4:4:4",
_ => "vulkan",
};
// Queue commands to send after the client loop.
pending_vulkan_encoder_setups.push(VulkanEncoderSetup {
surface_id: sid as u32,
codec: pref.vulkan_codec(),
qp,
width: px_w,
height: px_h,
});
pending_vulkan_keyframe_requests.push(sid as u32);
if let Some(s) = client.surface_subs.get_mut(&sid) {
s.encoder = None;
}
client
.vulkan_video_surfaces
.insert(sid, (enc_name, pref.codec_flag()));
let codec_str = match pref {
SurfaceEncoderPreference::VulkanVideoH264 => {
if state.config.chroma.is_444() {
"avc1.F4001f".to_string()
} else {
"avc1.640034".to_string()
}
}
SurfaceEncoderPreference::VulkanVideoAV1 => {
let profile = if state.config.chroma.is_444() { 2 } else { 0 };
let level = surface_encoder::av1_level_for(px_w, px_h);
format!("av01.{profile}.{level}M.08")
}
_ => String::new(),
};
let enc_msg = msg_surface_encoder(sid, enc_name, &codec_str);
let _ = send_outbox(client, enc_msg);
if state.config.verbose {
eprintln!(
"[surface-encoder] cid={} sid={sid} {px_w}x{px_h}: using {enc_name}",
work.cid,
);
}
break;
}
// Defer encoder creation to spawn_blocking so the
// tick loop isn't blocked by slow VA-API init.
// The creation task allocates GBM buffers and
// returns the encoder; the first encode runs on a
// subsequent tick, after the main loop forwards
// the buffers to the compositor and the compositor
// commits a new frame through them.
{
let state = client.surface_subs.entry(sid).or_default();
state.encoder = None;
state.creation_in_flight = true;
}
create_jobs.push(CreateJob {
cid: work.cid,
sid,
px_w: enc_w,
px_h: enc_h,
params: EncoderCreateParams {
preferences: state.config.surface_encoders.clone(),
vaapi_device: state.config.vaapi_device.clone(),
quality,
verbose: state.config.verbose,
codec_support,
chroma: state.config.chroma,
},
});
continue;
}
// If using Vulkan Video, handle keyframe via compositor
// command and skip local encode — the fast path above
// handles delivery.
if client.vulkan_video_surfaces.contains_key(&sid) {
if work.needs_keyframe {
pending_vulkan_keyframe_requests.push(sid as u32);
}
client.skip_vulkan_await_count =
client.skip_vulkan_await_count.saturating_add(1);
let now_inst = Instant::now();
if now_inst.duration_since(client.last_skip_log).as_secs_f32() > 5.0 {
client.last_skip_log = now_inst;
eprintln!(
"[encode-skip] cid={} sid={sid} reason=vulkan_await \
(compositor not producing PixelData::Encoded) count={}",
work.cid, client.skip_vulkan_await_count,
);
}
continue;
}
let encoder = client
.surface_subs
.get_mut(&sid)
.and_then(|s| s.encoder.take())
.unwrap();
client.surface_subs.entry(sid).or_default().encode_in_flight = true;
let needs_kf = work.needs_keyframe || needs_new_encoder;
encoded_client_surfaces.insert((work.cid, sid));
encode_dispatched_surfaces.insert(sid);
encode_jobs.push(EncodeJob {
cid: work.cid,
sid,
px_w: enc_w,
px_h: enc_h,
pixels,
needs_keyframe: needs_kf,
encoder,
generation: px_gen,
timestamp_ms: px_timestamp_ms,
});
}
}
// Send Vulkan Video encoder setup commands to compositor.
if (!pending_vulkan_encoder_setups.is_empty()
|| !pending_vulkan_keyframe_requests.is_empty())
&& let Some(cs) = sess.compositor.as_ref()
{
for setup in pending_vulkan_encoder_setups {
eprintln!(
"[vulkan-video] sending SetVulkanEncoder sid={} codec={} {}x{} qp={}",
setup.surface_id, setup.codec, setup.width, setup.height, setup.qp,
);
let _ = cs.handle.command_tx.send(
blit_compositor::CompositorCommand::SetVulkanEncoder {
surface_id: setup.surface_id,
codec: setup.codec,
qp: setup.qp,
width: setup.width,
height: setup.height,
},
);
}
for surface_id in pending_vulkan_keyframe_requests {
let _ = cs
.handle
.command_tx
.send(blit_compositor::CompositorCommand::RequestVulkanKeyframe { surface_id });
}
cs.handle.wake();
}
// Advance per-surface pacing deadlines only for surfaces that
// actually had an encode job collected. Surfaces skipped due to
// in-flight limits or unchanged pixels keep their current
// deadline so the next tick retries without burning a time slot.
for work in &client_work {
if let Some(client) = sess.clients.get_mut(&work.cid) {
let interval = surface_send_interval(client);
for &sid in &work.subs {
if encoded_client_surfaces.contains(&(work.cid, sid)) {
let deadline = client
.surface_subs
.entry(sid)
.or_default()
.next_send_at
.get_or_insert(now);
advance_deadline(deadline, now, interval);
}
}
}
}
}
if !encode_jobs.is_empty() {
// Fire-and-forget: spawn the encode and deliver asynchronously
// so the tick loop is never blocked by slow encoders.
let state2 = state.clone();
tokio::spawn(async move {
// Track (cid, sid) for each job so we can clear the sub's
// `encode_in_flight` flag if the blocking task panics or
// times out (otherwise that surface is permanently blocked).
let job_ids: Vec<(u64, u16)> = encode_jobs.iter().map(|j| (j.cid, j.sid)).collect();
let handles: Vec<_> = encode_jobs
.into_iter()
.map(|job| {
tokio::task::spawn_blocking(move || {
let mut encoder = job.encoder;
if job.needs_keyframe {
encoder.request_keyframe();
}
let nal_data = encoder.encode_pixels(&job.pixels);
let codec_flag = encoder.codec_flag();
EncodeResult {
cid: job.cid,
sid: job.sid,
px_w: job.px_w,
px_h: job.px_h,
generation: job.generation,
encoder,
nal_data,
codec_flag,
timestamp_ms: job.timestamp_ms,
}
})
})
.collect();
// Timeout: if a hardware encoder hangs (e.g. vaSyncSurface on
// AMD), don't block delivery of other surfaces' results forever.
const ENCODE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
let mut results = Vec::with_capacity(handles.len());
let mut failed: Vec<(u64, u16)> = Vec::new();
for (i, h) in handles.into_iter().enumerate() {
// Wrap the timeout in a nested tokio::spawn so that
// panics from tokio::time::timeout during runtime
// shutdown ("A Tokio 1.x context was found, but it is
// being shutdown") are caught as JoinErrors instead of
// crashing the outer task.
let wrapper =
tokio::spawn(async move { tokio::time::timeout(ENCODE_TIMEOUT, h).await });
match wrapper.await {
Ok(Ok(Ok(r))) => results.push(r),
Ok(Ok(Err(_join_err))) => {
// spawn_blocking panicked — encoder is lost.
let (cid, sid) = job_ids[i];
eprintln!("[surface-encoder] encode task panicked: cid={cid} sid={sid}",);
failed.push(job_ids[i]);
}
Ok(Err(_timeout)) => {
// Encoder hung (e.g. GPU hang in vaSyncSurface).
// The blocking thread is leaked but we must not
// let it stall all other surfaces forever.
let (cid, sid) = job_ids[i];
eprintln!(
"[surface-encoder] encode timed out ({}s): cid={cid} sid={sid}",
ENCODE_TIMEOUT.as_secs(),
);
failed.push(job_ids[i]);
}
Err(_join_err) => {
// Runtime shutting down — abandon remaining work.
eprintln!("[surface-encoder] runtime shutting down, aborting delivery");
return;
}
}
}
// Deliver encoded frames.
let mut sess = state2.session.lock().await;
let now = Instant::now();
let mut local_encodes = 0u32;
let mut local_encode_bytes = 0u64;
let mut local_frames_sent = 0u32;
// Clean up in-flight tracking for panicked/timed-out encodes.
// Without this, the surface is permanently blocked from
// future encode jobs and frame delivery stops for it.
for (cid, sid) in failed {
if let Some(client) = sess.clients.get_mut(&cid) {
if let Some(s) = client.surface_subs.get_mut(&sid) {
s.encode_in_flight = false;
}
// The encoder was moved into the spawn_blocking closure
// and is now lost. A fresh encoder will be created on
// the next tick when the sub's encoder is None. Force
// a keyframe so the new encoder starts with a clean
// reference chain.
client.surface_needs_keyframe = true;
}
}
for result in results {
// Return the encoder to the client, but only if its
// dimensions still match the current surface. A resize
// that arrived while the encode was in flight will have
// invalidated the old encoder; reinserting the stale one
// would force the next tick to discard and recreate it,
// wasting work and risking feeding a C encoder (openh264)
// frames at the wrong resolution.
let expected_dims: Option<(u32, u32)> = sess
.compositor
.as_ref()
.and_then(|cs| cs.last_pixels.get(&result.sid))
.map(|lp| (lp.width, lp.height));
let dims_match =
expected_dims.is_some_and(|d| result.encoder.source_dimensions() == d);
if let Some(client) = sess.clients.get_mut(&result.cid) {
let state = client.surface_subs.entry(result.sid).or_default();
state.encode_in_flight = false;
let invalidated = std::mem::replace(&mut state.encoder_invalidated, false);
if dims_match && !invalidated {
state.encoder = Some(result.encoder);
}
// Record the generation we just encoded so we don't
// re-encode identical pixel data on subsequent ticks.
state.last_encoded_gen = Some(result.generation);
}
let Some((nal_data, is_keyframe)) = result.nal_data else {
if let Some(client) = sess.clients.get_mut(&result.cid) {
let state = client.surface_subs.entry(result.sid).or_default();
state.nal_none_streak += 1;
let streak = state.nal_none_streak;
if streak == 10 {
state.encoder = None;
state.nal_none_latched_at = Some(now);
client.surface_needs_keyframe = true;
eprintln!(
"[encode] nal_data=None x{streak} sid={} cid={} {}x{} — dropping encoder, backing off retry",
result.sid, result.cid, result.px_w, result.px_h,
);
} else if streak < 10 {
eprintln!(
"[encode] nal_data=None sid={} cid={} {}x{}",
result.sid, result.cid, result.px_w, result.px_h,
);
}
// streak >= 10: suppress the log spam
}
continue;
};
// Encoder produced output — reset the None streak.
if let Some(client) = sess.clients.get_mut(&result.cid)
&& let Some(s) = client.surface_subs.get_mut(&result.sid)
{
s.nal_none_streak = 0;
}
{
static EC: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let n = EC.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if n < 5 || n.is_multiple_of(1000) {
eprintln!(
"[encode #{n}] sid={} {}x{} kf={is_keyframe} bytes={}",
result.sid,
result.px_w,
result.px_h,
nal_data.len(),
);
}
}
local_encodes += 1;
local_encode_bytes += nal_data.len() as u64;
let flags = result.codec_flag
| if is_keyframe {
SURFACE_FRAME_FLAG_KEYFRAME
} else {
0
};
let msg = msg_surface_frame(
result.sid,
result.timestamp_ms,
flags,
result.px_w as u16,
result.px_h as u16,
&nal_data,
);
let bytes = msg.len();
let Some(client) = sess.clients.get_mut(&result.cid) else {
continue;
};
// Don't check window_open here — we already checked before
// starting the encode job. Dropping an encoded P-frame
// breaks the decoder's reference chain and causes glitches.
// With the per-sub `encode_in_flight` flag limiting to 1
// concurrent encode per surface, at most 1 frame arrives
// after the window closes, which is acceptable.
match send_outbox(client, msg) {
Err(_e) => {
// Receiver dropped (client disconnected during encode).
// Request keyframe so the next encoder starts clean.
client.surface_needs_keyframe = true;
}
Ok(()) => {
// Track surface frames in their own inflight queue
// so surface ACKs feed shared goodput / RTT without
// polluting terminal frame-size averages or probing.
client.surface_inflight_frames.push_back(InFlightFrame {
sent_at: now,
bytes,
paced: true,
});
// Prefer updating avg_surface_frame_bytes from delta
// (non-keyframe) frames — keyframes are 5-10× larger
// than P-frames and would inflate the average, dragging
// surface_pacing_fps below the sustainable rate.
//
// However, we must still update from keyframes with a
// very slow alpha: all-intra encoders (e.g. AV1 VAAPI
// before P-frame support) only produce keyframes, so
// skipping them entirely leaves the average stuck at
// the 8 KB initial value, causing the pacer to wildly
// overshoot the send rate and saturate the transport.
if !is_keyframe {
client.avg_surface_frame_bytes = ewma_with_direction(
client.avg_surface_frame_bytes,
bytes as f32,
0.5,
0.125,
);
} else if client.avg_surface_frame_bytes <= 16_384.0 {
// First keyframe while the estimate is still at or
// near the initial 8 KB seed. No P-frame data has
// been seen yet, so the seed is pure fiction. Use a
// realistic P-frame estimate: keyframes are typically
// 3-8× larger than P-frames, so divide by 4. This
// prevents surface_pacing_fps from being wildly
// optimistic (8 KB → 32 fps at 256 KB/s) when the
// actual frames are 50-200 KB keyframes.
client.avg_surface_frame_bytes = (bytes as f32 / 4.0).max(4_096.0);
} else {
// Slow convergence so one keyframe doesn't wreck
// the estimate for dozens of subsequent P-frames.
client.avg_surface_frame_bytes = ewma_with_direction(
client.avg_surface_frame_bytes,
bytes as f32,
0.05,
0.05,
);
}
client.frames_sent = client.frames_sent.wrapping_add(1);
local_frames_sent += 1;
if client.surface_needs_keyframe && is_keyframe {
client.surface_needs_keyframe = false;
}
if let Some(s) = client.surface_subs.get_mut(&result.sid) {
s.burst_remaining = s.burst_remaining.saturating_sub(1);
}
}
}
}
sess.surface_encodes += local_encodes;
sess.surface_encode_bytes += local_encode_bytes;
sess.surface_frames_sent += local_frames_sent;
drop(sess);
// Wake the tick loop so it can request the next frame.
state2.delivery_notify.notify_one();
});
}
if !create_jobs.is_empty() {
// Encoder creation runs on spawn_blocking so VA-API device open
// and context allocation don't stall the tick loop. When the
// task lands, the main loop installs the encoder into the sub's
// `encoder` slot, forwards the GBM buffers to the compositor
// (`SetExternalOutputBuffers`), and sends S2C_SURFACE_ENCODER
// to the client. Encoding starts on the NEXT tick — once the
// compositor has committed a frame through the new buffers.
let state2 = state.clone();
tokio::spawn(async move {
// Track (cid, sid) for each job so we can clear
// `creation_in_flight` if a task panics or times out.
let job_ids: Vec<(u64, u16)> = create_jobs.iter().map(|j| (j.cid, j.sid)).collect();
let handles: Vec<_> = create_jobs
.into_iter()
.map(|job| {
tokio::task::spawn_blocking(move || {
let params = job.params;
let mut encoder = match SurfaceEncoder::new(
¶ms.preferences,
job.px_w,
job.px_h,
¶ms.vaapi_device,
params.quality,
params.verbose,
params.codec_support,
params.chroma,
) {
Ok(enc) => enc,
Err(err) => {
if params.verbose {
eprintln!(
"[surface-encoder] cid={} sid={} {}x{}: {err}",
job.cid, job.sid, job.px_w, job.px_h,
);
}
return CreateResult {
cid: job.cid,
sid: job.sid,
encoder: None,
fresh: None,
};
}
};
#[cfg(target_os = "linux")]
let external_bufs = {
{
let drm_fd = encoder.drm_fd_raw();
let count = encoder.gbm_buffers().len();
if count > 0 {
encoder.allocate_nv12_buffers(drm_fd, count);
}
}
let gbm_bufs = encoder.gbm_buffers();
if gbm_bufs.is_empty() {
Vec::new()
} else {
let nv12_bufs = encoder.gbm_nv12_buffers();
let (enc_w, enc_h) = encoder.encoder_dimensions();
let bufs: Result<Vec<_>, std::io::Error> = gbm_bufs
.iter()
.enumerate()
.map(|(i, b)| {
let nv12 = nv12_bufs.get(i);
Ok(blit_compositor::ExternalOutputBuffer {
fd: std::sync::Arc::new(b.fd.try_clone()?),
fourcc: 0x34325241,
modifier: 0,
stride: b.stride,
offset: 0,
width: b.width,
height: b.height,
va_surface_id: 0,
va_display: 0,
planes: vec![blit_compositor::ExternalOutputPlane {
offset: 0,
pitch: b.stride,
}],
nv12_fd: nv12.map(|n| n.fd.clone()),
nv12_stride: nv12.map_or(0, |n| n.stride),
nv12_uv_offset: nv12.map_or(0, |n| n.uv_offset),
nv12_modifier: nv12.map_or(0, |n| n.modifier),
nv12_width: enc_w,
nv12_height: enc_h,
})
})
.collect();
match bufs {
Ok(b) => b,
Err(e) => {
eprintln!("[encode] dup gbm fd failed: {e}");
Vec::new()
}
}
}
};
let fresh = FreshEncoder {
name: encoder.encoder_name(),
codec_string: encoder.webcodecs_codec_string(),
#[cfg(target_os = "linux")]
external_bufs,
};
CreateResult {
cid: job.cid,
sid: job.sid,
encoder: Some(encoder),
fresh: Some(fresh),
}
})
})
.collect();
const CREATE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
let mut results: Vec<CreateResult> = Vec::with_capacity(handles.len());
let mut failed: Vec<(u64, u16)> = Vec::new();
for (i, h) in handles.into_iter().enumerate() {
let wrapper =
tokio::spawn(async move { tokio::time::timeout(CREATE_TIMEOUT, h).await });
match wrapper.await {
Ok(Ok(Ok(r))) => results.push(r),
Ok(Ok(Err(_))) | Ok(Err(_)) => {
let (cid, sid) = job_ids[i];
eprintln!("[surface-encoder] create task failed: cid={cid} sid={sid}",);
failed.push(job_ids[i]);
}
Err(_) => return,
}
}
let mut sess = state2.session.lock().await;
let now = Instant::now();
// Clear creation_in_flight for failed tasks; latch a brief
// backoff so the next tick doesn't immediately retry.
for (cid, sid) in failed {
if let Some(client) = sess.clients.get_mut(&cid)
&& let Some(s) = client.surface_subs.get_mut(&sid)
{
s.creation_in_flight = false;
s.nal_none_streak = 10;
s.nal_none_latched_at = Some(now);
}
}
for result in results {
let Some(encoder) = result.encoder else {
if let Some(client) = sess.clients.get_mut(&result.cid)
&& let Some(s) = client.surface_subs.get_mut(&result.sid)
{
s.creation_in_flight = false;
s.nal_none_streak = 10;
s.nal_none_latched_at = Some(now);
}
continue;
};
// Move the external buffers (and register them with the
// compositor) BEFORE stashing the encoder, so subsequent
// ticks see the encoder only once its buffers are live.
let fresh = result.fresh;
#[cfg(target_os = "linux")]
{
if let Some(f) = &fresh
&& !f.external_bufs.is_empty()
&& let Some(cs) = sess.compositor.as_mut()
{
cs.last_pixels.remove(&result.sid);
}
}
#[cfg(target_os = "linux")]
let (fresh_meta, external_bufs) = match fresh {
Some(f) => (Some((f.name, f.codec_string)), Some(f.external_bufs)),
None => (None, None),
};
#[cfg(not(target_os = "linux"))]
let fresh_meta = fresh.map(|f| (f.name, f.codec_string));
#[cfg(target_os = "linux")]
if let Some(bufs) = external_bufs
&& !bufs.is_empty()
&& let Some(cs) = sess.compositor.as_mut()
{
let _ = cs.handle.command_tx.send(
blit_compositor::CompositorCommand::SetExternalOutputBuffers {
surface_id: result.sid as u32,
buffers: bufs,
},
);
cs.handle.wake();
}
if let Some(client) = sess.clients.get_mut(&result.cid) {
let state = client.surface_subs.entry(result.sid).or_default();
state.creation_in_flight = false;
let invalidated = std::mem::replace(&mut state.encoder_invalidated, false);
if invalidated {
// Preferences changed mid-creation (codec/quality
// resubscribe). Drop the encoder we just built;
// the next tick will dispatch a fresh creation
// with the new prefs.
continue;
}
state.encoder = Some(encoder);
state.nal_none_streak = 0;
state.nal_none_latched_at = None;
if let Some((name, codec_string)) = fresh_meta {
let enc_msg = msg_surface_encoder(result.sid, name, &codec_string);
let _ = send_outbox(client, enc_msg);
}
}
}
drop(sess);
state2.delivery_notify.notify_one();
});
}
// Request frames from the compositor for surfaces that have at least
// one subscriber whose pacing says it can accept a new frame. This
// fires the surface's pending wl_surface.frame callback so the
// Wayland client will paint and commit its next frame.
//
// Demand-driven with pipeline overlap:
// When an encode job is dispatched, we eagerly pre-request the next
// frame so the Wayland client paints in parallel with the encode.
// Fresh pixels are ready when the encode completes, turning the
// serial encode + round_trip into max(encode, round_trip).
{
// Only request frames for surfaces where at least one client is
// ready to consume the result. Without this check, apps that are
// always ready to paint (video players like mpv) cause a hot loop:
// RequestFrame → commit → SurfaceCommit wakes tick → no client
// ready → RequestFrame again → 100% CPU.
let mut wanted: HashSet<u16> = HashSet::new();
// Pre-request: surfaces with an encode just dispatched. The
// compositor will render the next frame while the encode runs,
// so pixels are ready when the next pacing window opens.
for &sid in &encode_dispatched_surfaces {
wanted.insert(sid);
}
let mut blanket_requested = false;
// Request frames for all known surfaces so Wayland apps can make
// rendering progress. Video players (mpv) need frequent callbacks
// to advance their presentation clock; browsers need them for
// page loads and animations.
if let Some(cs) = sess.compositor.as_ref()
&& let Some(interval) = blanket_frame_interval(&sess)
&& now.duration_since(cs.last_blanket_frame_request) >= interval
{
for &sid in cs.surfaces.keys() {
wanted.insert(sid);
}
blanket_requested = true;
}
for client in sess.clients.values() {
// Don't gate frame requests on surface_window_open — the
// compositor should keep producing pixels even when the
// inflight window is closed. Otherwise, recovery after a
// wifi stall has to wait for the full render pipeline to
// flush (request → paint → commit → encode) before the
// first frame can be sent, causing a visible hang.
if client.surface_subscriptions.is_empty() {
continue;
}
for &sid in &client.surface_subscriptions {
let (burst, deadline) = client.surface_subs.get(&sid).map_or((0, now), |s| {
(s.burst_remaining, s.next_send_at.unwrap_or(now))
});
if deadline <= now || burst > 0 {
wanted.insert(sid);
} else {
next_deadline = Some(match next_deadline {
Some(existing) => existing.min(deadline),
None => deadline,
});
}
}
}
if let Some(cs) = sess.compositor.as_mut() {
if blanket_requested {
cs.last_blanket_frame_request = now;
}
// Gate: at most one RequestFrame per surface per millisecond.
// This ensures each wl_callback.done carries a distinct
// elapsed_ms timestamp (video players like mpv use these to
// pace their presentation clock). Supports up to 1 kHz.
// The gate auto-expires: if the app doesn't commit, the next
// tick ≥1 ms later will send a fresh request.
const MIN_REQUEST_INTERVAL: Duration = Duration::from_millis(1);
let mut sent_any = false;
for sid in &wanted {
let dominated = cs
.last_frame_request
.get(sid)
.is_some_and(|&t| now.duration_since(t) < MIN_REQUEST_INTERVAL);
if !dominated {
cs.last_frame_request.insert(*sid, now);
let _ = cs
.handle
.command_tx
.send(CompositorCommand::RequestFrame { surface_id: *sid });
sent_any = true;
}
}
if sent_any {
cs.handle.wake();
}
}
}
// Yield the session lock briefly so pending encode deliveries from
// previous ticks can acquire the lock and send their frames without
// waiting for terminal processing to complete. This reduces the
// latency between encode completion and frame-on-wire.
drop(sess);
tokio::task::yield_now().await;
sess = state.session.lock().await;
let max_fps = sess
.clients
.values()
.map(browser_pacing_fps)
.fold(1.0_f32, f32::max);
let title_interval = Duration::from_secs_f64(1.0 / max_fps as f64);
let ids: Vec<u16> = sess.ptys.keys().copied().collect();
for &id in &ids {
let Some(pty) = sess.ptys.get_mut(&id) else {
continue;
};
if pty.driver.take_title_dirty() {
pty.mark_dirty();
pty.title_pending = true;
}
if pty.title_pending && now.duration_since(pty.last_title_send) >= title_interval {
let msg = {
let title_bytes = pty.driver.title().as_bytes();
let mut msg = Vec::with_capacity(3 + title_bytes.len());
msg.push(S2C_TITLE);
msg.extend_from_slice(&id.to_le_bytes());
msg.extend_from_slice(title_bytes);
msg
};
pty.last_title_send = now;
pty.title_pending = false;
sess.send_to_all(&msg);
}
}
// Drain bytes from PTY reader channels. This is the only place
// process() is called, so there is no contention with the readers.
//
// End-to-end flow control: when at least one client is subscribed to
// a PTY and its `ready_frames` queue is full, stop draining `byte_rx`
// for that PTY. `byte_rx` then fills to its bounded capacity, which
// blocks the reader task's `byte_tx.blocking_send`, which fills the
// kernel's PTY master buffer, which blocks the child process's
// `write(stdout, ...)`. Sync-bracketed frames are never silently
// dropped; the producer is slowed instead. PTYs with no subscribers
// drain unconditionally so background processes aren't throttled by
// nobody-watching.
let ptys_with_subscribers: HashSet<u16> = sess
.clients
.values()
.flat_map(|c| c.subscriptions.iter().copied())
.collect();
let mut eof_ptys: Vec<u16> = Vec::with_capacity(ids.len());
for &id in &ids {
let Some(pty) = sess.ptys.get_mut(&id) else {
continue;
};
let has_subscriber = ptys_with_subscribers.contains(&id);
loop {
if has_subscriber && pty.ready_frames.len() >= READY_FRAME_QUEUE_CAP {
break;
}
let Ok(input) = pty.byte_rx.try_recv() else {
break;
};
match input {
PtyInput::Data(data) => {
pty::respond_to_queries(
&pty.handle,
&data,
pty.driver.size(),
pty.driver.cursor_position(),
);
pty.driver.process(&data);
pty.mark_dirty();
}
PtyInput::SyncBoundary { before } => {
if !before.is_empty() {
pty::respond_to_queries(
&pty.handle,
&before,
pty.driver.size(),
pty.driver.cursor_position(),
);
pty.driver.process(&before);
pty.mark_dirty();
}
if !pty.driver.synced_output() {
let frame = take_snapshot(pty);
enqueue_ready_frame(&mut pty.ready_frames, frame);
pty.clear_dirty();
}
}
PtyInput::Eof => {
eof_ptys.push(id);
}
}
}
}
// Handle EOF outside the borrow loop.
drop(sess);
for id in eof_ptys {
tokio::time::sleep(Duration::from_millis(50)).await;
cleanup_pty_internal(id, state).await;
}
let mut sess = state.session.lock().await;
// Only snapshot PTYs that have at least one client ready to consume a fresh
// frame right now. This avoids burning CPU on snapshot+diff+compress work
// while the lead is merely waiting for its next pacing deadline.
let needful_ptys: HashSet<u16> = sess
.clients
.values()
.flat_map(|c| {
let reserve_preview_slot = client_has_due_preview(&sess, c, now);
c.subscriptions.iter().copied().filter(move |pid| {
let scrolled = c.scroll_offsets.get(pid).copied().unwrap_or(0) > 0;
if Some(*pid) == c.lead {
!scrolled && can_send_frame(c, now, reserve_preview_slot)
} else {
!scrolled && can_send_preview(c, *pid, now)
}
})
})
.collect();
let mut snapshots: HashMap<u16, FrameState> = HashMap::new();
for &id in &ids {
let Some(pty) = sess.ptys.get_mut(&id) else {
continue;
};
if needful_ptys.contains(&id)
&& let Some(frame) = pty.ready_frames.pop_front()
{
snapshots.insert(id, frame);
sess.tick_snaps += 1;
continue;
}
if !should_snapshot_pty(
pty.dirty,
needful_ptys.contains(&id),
pty.driver.synced_output(),
) {
continue;
}
// Applications that care about complete-frame boundaries should
// use DEC synchronized output (?2026). Outside that bracket we
// snapshot immediately instead of heuristically coalescing reads.
snapshots.insert(id, take_snapshot(pty));
pty.clear_dirty();
sess.tick_snaps += 1;
}
let client_ids: Vec<u64> = sess.clients.keys().copied().collect();
for cid in client_ids {
// When the pipe is idle (nothing in flight), RTT cannot be measured
// and the last observed value stales. Decay it toward min_rtt so
// a stale congested RTT doesn't permanently suppress the send window
// after congestion clears or traffic patterns change (e.g. switching
// from a large-frame burst to idle small-frame updates).
if let Some(c) = sess.clients.get_mut(&cid) {
if c.inflight_bytes == 0 && c.min_rtt_ms > 0.0 && c.rtt_ms > c.min_rtt_ms {
c.rtt_ms = (c.rtt_ms * 0.99 + c.min_rtt_ms * 0.01).max(c.min_rtt_ms);
}
// Decay stale browser metrics so a missed/delayed metrics update
// can't permanently block the delivery loop.
if c.last_metrics_update.elapsed() > Duration::from_secs(1) {
c.browser_backlog_frames = 0;
c.browser_ack_ahead_frames = 0;
}
}
let (
lead,
subscriptions,
scrolled_ptys,
can_send_lead,
lead_has_window,
any_send_window,
lead_deadline,
) = {
let Some(c) = sess.clients.get(&cid) else {
continue;
};
let reserve_preview_slot = client_has_due_preview(&sess, c, now);
(
c.lead,
c.subscriptions.iter().copied().collect::<Vec<_>>(),
c.scroll_offsets
.iter()
.map(|(&k, &v)| (k, v))
.collect::<Vec<_>>(),
can_send_frame(c, now, reserve_preview_slot),
lead_window_open(c, reserve_preview_slot),
lead_window_open(c, reserve_preview_slot) || window_open(c),
c.next_send_at,
)
};
if subscriptions.is_empty() {
continue;
}
// Send scrollback frames for any scrolled PTY.
for &(scroll_pid, scroll_offset) in &scrolled_ptys {
if scroll_offset == 0 {
continue;
}
let is_lead = lead == Some(scroll_pid);
let can_send = if is_lead { can_send_lead } else { true };
if can_send {
let prev_frame = {
let Some(c) = sess.clients.get(&cid) else {
continue;
};
c.scroll_caches
.get(&scroll_pid)
.cloned()
.unwrap_or_default()
};
let outcome = if let Some(pty) = sess.ptys.get_mut(&scroll_pid) {
if let Some((msg, new_frame)) =
build_scrollback_update(pty, scroll_pid, scroll_offset, &prev_frame)
{
let Some(c) = sess.clients.get_mut(&cid) else {
break;
};
let bytes = msg.len();
if send_outbox(c, msg).is_ok() {
c.scroll_caches.insert(scroll_pid, new_frame);
record_send(c, bytes, now, is_lead);
c.frames_sent += 1;
SendOutcome::Sent
} else {
SendOutcome::Backpressured
}
} else {
SendOutcome::NoChange
}
} else {
SendOutcome::NoChange
};
match outcome {
SendOutcome::Sent => {}
SendOutcome::Backpressured => {
if let Some(pty) = sess.ptys.get_mut(&scroll_pid) {
pty.mark_dirty();
}
}
SendOutcome::NoChange => {}
}
} else if is_lead && lead_has_window {
next_deadline = Some(match next_deadline {
Some(existing) => existing.min(lead_deadline),
None => lead_deadline,
});
}
}
let lead_scroll_offset = lead
.and_then(|pid| {
scrolled_ptys
.iter()
.find(|&&(k, _)| k == pid)
.map(|&(_, v)| v)
})
.unwrap_or(0);
if let Some(pid) = lead {
if lead_scroll_offset == 0 && can_send_lead {
if let Some(cur) = snapshots.get(&pid).cloned() {
let previous = sess
.clients
.get(&cid)
.and_then(|c| c.last_sent.get(&pid).cloned())
.unwrap_or_default();
drop(sess);
let msg = build_update_msg(pid, &cur, &previous);
sess = state.session.lock().await;
let Some(c) = sess.clients.get_mut(&cid) else {
continue;
};
match try_send_update(c, pid, cur, msg, now, true) {
SendOutcome::Sent => {}
SendOutcome::Backpressured => {
if let Some(pty) = sess.ptys.get_mut(&pid) {
pty.mark_dirty();
}
}
SendOutcome::NoChange => {}
}
} else {
let has_pending = sess
.ptys
.get(&pid)
.map(pty_has_visual_update)
.unwrap_or(false);
let _ = has_pending;
}
} else {
let has_pending = sess
.ptys
.get(&pid)
.map(pty_has_visual_update)
.unwrap_or(false);
if has_pending && lead_has_window {
next_deadline = Some(match next_deadline {
Some(existing) => existing.min(lead_deadline),
None => lead_deadline,
});
}
}
}
if !any_send_window {
continue;
}
let mut preview_ids = subscriptions;
preview_ids.retain(|pid| Some(*pid) != lead);
preview_ids.sort_unstable();
for pid in preview_ids {
let (preview_can_send, preview_due_at, preview_has_window) =
match sess.clients.get(&cid) {
Some(c) => (
can_send_preview(c, pid, now),
preview_deadline(c, pid, now),
window_open(c),
),
None => (false, now, false),
};
if !preview_has_window {
break;
}
if !preview_can_send {
let has_pending = sess
.ptys
.get(&pid)
.map(pty_has_visual_update)
.unwrap_or(false);
// Only set a deadline when the reason is *timing* (deadline
// in the future), not capacity (preview window closed).
// A past deadline here spins the delivery loop because
// sleep_until(past) returns immediately.
if has_pending && preview_due_at > now {
next_deadline = Some(match next_deadline {
Some(existing) => existing.min(preview_due_at),
None => preview_due_at,
});
}
continue;
}
let Some(cur) = snapshots.get(&pid) else {
let has_pending = sess
.ptys
.get(&pid)
.map(pty_has_visual_update)
.unwrap_or(false);
let _ = has_pending;
continue;
};
let cur = cur.clone();
let previous = sess
.clients
.get(&cid)
.and_then(|c| c.last_sent.get(&pid).cloned())
.unwrap_or_default();
drop(sess);
let msg = build_update_msg(pid, &cur, &previous);
sess = state.session.lock().await;
let Some(c) = sess.clients.get_mut(&cid) else {
break;
};
match try_send_update(c, pid, cur, msg, now, false) {
SendOutcome::Sent => {
record_preview_send(c, pid, now);
}
SendOutcome::Backpressured => {
if let Some(pty) = sess.ptys.get_mut(&pid) {
pty.mark_dirty();
}
break;
}
SendOutcome::NoChange => {}
}
}
}
// -- Audio frame delivery -----------------------------------------------
//
// Audio is no longer delivered from the tick loop — a dedicated
// fan-out task (spawned in `AudioPipeline::spawn`) drains encoded
// frames from the encoder mpsc and pushes them to each subscribed
// client's `audio_tx` independently of compositor/video work. This
// keeps audio flowing at a steady 20 ms cadence even when a tick is
// blocked by a long video write, and keeps the encoder's bounded
// mpsc from overflowing into silent frame drops.
//
// Audio bytes are intentionally excluded from `goodput_window_bytes`:
// at ~8 KB/s they're negligible next to video (MB/s) and keeping the
// accounting on the tick loop would defeat the whole point of the
// off-tick fan-out. The has_listener flag is now managed by the
// subscribe/unsubscribe API on `AudioBroadcast`.
// -- Audio pipeline auto-restart ----------------------------------------
// If the pipeline died (encoder crashed, PipeWire gone, capture stream dropped),
// drop it, wait for a cooldown, and respawn. This avoids permanent
// audio loss that previously required a full client reconnect.
//
// Bitrate is pre-computed here to avoid borrowing sess.clients inside
// the sess.compositor mutable borrow (they're the same MutexGuard).
#[cfg(target_os = "linux")]
let audio_restart_bitrate: i32 = sess
.clients
.values()
.filter(|c| c.audio_subscribed)
.map(|c| c.audio_bitrate_kbps)
.max()
.map(|kbps| kbps as i32 * 1000)
.unwrap_or(0);
#[cfg(target_os = "linux")]
if let Some(ref mut cs) = sess.compositor {
let pipeline_dead = cs.audio_pipeline.as_mut().is_some_and(|ap| !ap.is_alive());
if pipeline_dead {
const RESTART_COOLDOWN: Duration = Duration::from_secs(5);
let can_restart = cs
.last_audio_restart
.is_none_or(|t| now.duration_since(t) >= RESTART_COOLDOWN);
if can_restart {
cs.last_audio_restart = Some(now);
// Drop the dead pipeline — triggers shutdown() which kills
// orphaned child processes and cleans up the runtime dir.
cs.audio_pipeline = None;
let runtime_dir = std::path::Path::new(&cs.handle.socket_name)
.parent()
.unwrap_or(std::path::Path::new("/tmp"));
let session_id = cs.audio_session_id;
let epoch = cs.created_at;
let verbose = state.config.verbose;
// Reuse the existing broadcast so currently-subscribed
// clients pick up frames from the restarted pipeline
// without re-subscribing.
let broadcast = cs.audio_broadcast.clone();
eprintln!("[audio] pipeline died, restarting...");
let pipeline = tokio::task::block_in_place(|| {
audio::AudioPipeline::spawn(
runtime_dir,
session_id,
audio_restart_bitrate,
verbose,
epoch,
broadcast,
)
});
match pipeline {
Ok(p) => {
eprintln!(
"[audio] pipeline restarted, PULSE_SERVER={}",
p.pulse_server_path(),
);
cs.audio_pipeline = Some(p);
}
Err(e) => {
eprintln!("[audio] failed to restart pipeline: {e}");
}
}
}
}
}
// Guarantee the tick loop wakes up at least every blanket interval
// even when other time-based work isn't pending. When no client is
// connected the interval is `None` and the loop sleeps purely on
// delivery_notify, so a truly-idle server consumes ~zero CPU until
// a client connects or the compositor emits an event.
if let Some(interval) = blanket_frame_interval(&sess) {
let blanket_deadline = now + interval;
next_deadline = Some(next_deadline.map_or(blanket_deadline, |d| d.min(blanket_deadline)));
}
TickOutcome { next_deadline }
}
async fn handle_client<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
stream: S,
state: AppState,
) {
let config = &state.config;
let notify_for_compositor = {
let n = state.delivery_notify.clone();
Arc::new(move || n.notify_one()) as Arc<dyn Fn() + Send + Sync>
};
let (mut reader, mut writer) = tokio::io::split(stream);
let (out_tx, mut out_rx) = mpsc::unbounded_channel::<Vec<u8>>();
let (audio_tx, mut audio_rx) = mpsc::unbounded_channel::<Vec<u8>>();
let outbox_frame_counter = Arc::new(AtomicUsize::new(0));
let outbox_byte_counter = Arc::new(AtomicUsize::new(0));
let sender_outbox_queued_frames = outbox_frame_counter.clone();
let sender_outbox_queued_bytes = outbox_byte_counter.clone();
let sender = tokio::spawn(async move {
let audio_debug = std::env::var_os("BLIT_AUDIO_DEBUG").is_some();
let mut audio_window_start = Instant::now();
let mut last_audio_pick_at = Instant::now();
let mut audio_sends_in_window: u32 = 0;
let mut max_audio_pick_gap: u32 = 0;
let mut max_audio_write_ms: u32 = 0;
loop {
// Drain all pending audio before waiting for the next message.
// Audio frames are tiny (~160 B) so this is near-instant.
while let Ok(audio_msg) = audio_rx.try_recv() {
if !write_frame(&mut writer, &audio_msg).await {
return;
}
if audio_debug {
audio_sends_in_window += 1;
let now = Instant::now();
let pick_gap = now.duration_since(last_audio_pick_at).as_millis() as u32;
last_audio_pick_at = now;
if pick_gap > max_audio_pick_gap {
max_audio_pick_gap = pick_gap;
}
}
}
// Wait for the next message from either channel. Prefer audio
// so that audio frames queued while we were writing are sent
// before the next bulk message.
let msg = tokio::select! {
biased;
msg = audio_rx.recv() => {
// Pure audio message — write it directly (tiny).
match msg {
Some(m) => {
let audio_write_start = Instant::now();
if !write_frame(&mut writer, &m).await {
break;
}
if audio_debug {
let now = Instant::now();
audio_sends_in_window += 1;
let pick_gap = now
.duration_since(last_audio_pick_at)
.as_millis() as u32;
last_audio_pick_at = now;
let write_ms =
now.duration_since(audio_write_start).as_millis() as u32;
if pick_gap > max_audio_pick_gap {
max_audio_pick_gap = pick_gap;
}
if write_ms > max_audio_write_ms {
max_audio_write_ms = write_ms;
}
if now.duration_since(audio_window_start)
>= Duration::from_secs(1)
{
eprintln!(
"[sender audio] writes={} max_pick_gap={}ms max_write={}ms",
audio_sends_in_window,
max_audio_pick_gap,
max_audio_write_ms,
);
audio_sends_in_window = 0;
max_audio_pick_gap = 0;
max_audio_write_ms = 0;
audio_window_start = now;
}
}
continue;
}
None => break,
}
}
msg = out_rx.recv() => msg,
};
// Non-audio message: may be large (video keyframe, terminal
// snapshot). Use interleaved write so audio frames that arrive
// while the kernel TCP buffer drains are written between write
// syscalls rather than piling up and being dropped.
match msg {
Some(m) => {
let bytes = m.len();
let write_start = Instant::now();
let wrote = write_frame_interleaved(&mut writer, &m, &mut audio_rx).await;
let write_elapsed = write_start.elapsed();
// Threshold lowered from 100 ms to 30 ms so sub-chunk
// stalls on slow links (the band that can still
// block audio delivery for longer than the 20 ms
// Opus frame cadence) show up in the log.
if write_elapsed.as_millis() > 30 {
eprintln!(
"[sender] slow write: bytes={bytes} elapsed={}ms wrote={wrote}",
write_elapsed.as_millis(),
);
}
mark_outbox_drained(
&sender_outbox_queued_frames,
&sender_outbox_queued_bytes,
bytes,
);
if !wrote {
break;
}
}
None => break,
}
}
});
let client_id;
{
let mut sess = state.session.lock().await;
client_id = sess.next_client_id;
sess.next_client_id += 1;
sess.clients.insert(
client_id,
ClientState {
tx: out_tx,
outbox_queued_frames: outbox_frame_counter,
outbox_queued_bytes: outbox_byte_counter,
audio_tx,
lead: None,
subscriptions: HashSet::new(),
surface_subscriptions: HashSet::new(),
audio_subscribed: false,
#[cfg(target_os = "linux")]
audio_bitrate_kbps: 0,
view_sizes: HashMap::new(),
scroll_offsets: HashMap::new(),
scroll_caches: HashMap::new(),
last_sent: HashMap::new(),
preview_next_send_at: HashMap::new(),
rtt_ms: 50.0,
min_rtt_ms: 0.0,
display_fps: 60.0,
// Conservative seed — the rise alpha (0.5) converges up to
// multi-MB/s in a handful of samples on low-latency paths. Starting
// high causes catastrophic bufferbloat on slow links because
// target_byte_window scales with the goodput estimate.
delivery_bps: 262_144.0,
goodput_bps: 262_144.0,
goodput_jitter_bps: 0.0,
max_goodput_jitter_bps: 0.0,
last_goodput_sample_bps: 0.0,
avg_frame_bytes: 1_024.0,
avg_paced_frame_bytes: 1_024.0,
avg_preview_frame_bytes: 1_024.0,
avg_surface_frame_bytes: 8_192.0,
inflight_bytes: 0,
inflight_frames: VecDeque::new(),
next_send_at: Instant::now(),
probe_frames: 0.0,
frames_sent: 0,
acks_recv: 0,
acked_bytes_since_log: 0,
browser_backlog_frames: 0,
browser_ack_ahead_frames: 0,
browser_apply_ms: 0.0,
last_metrics_update: Instant::now(),
last_log: Instant::now(),
last_window_blocked_log: Instant::now(),
last_skip_log: Instant::now(),
skip_same_gen_count: 0,
skip_in_flight_count: 0,
skip_pacing_count: 0,
skip_vulkan_await_count: 0,
skip_no_subs_count: 0,
skip_not_subbed_count: 0,
skip_last_pixels_mismatch_count: 0,
encode_loop_iters: 0,
goodput_window_bytes: 0,
goodput_window_start: Instant::now(),
surface_subs: HashMap::new(),
surface_needs_keyframe: true,
surface_inflight_frames: VecDeque::new(),
vulkan_video_surfaces: HashMap::new(),
surface_view_sizes: HashMap::new(),
surface_codec_support: 0,
pressed_surface_keys: HashSet::new(),
},
);
// Wake the tick loop so the new client gets its first frame.
state.delivery_notify.notify_one();
if let Some(c) = sess.clients.get(&client_id) {
let mut features = FEATURE_CREATE_NONCE
| FEATURE_RESTART
| FEATURE_RESIZE_BATCH
| FEATURE_COPY_RANGE
| FEATURE_COMPOSITOR;
#[cfg(target_os = "linux")]
{
let audio_disabled = std::env::var("BLIT_AUDIO")
.map(|v| v == "0")
.unwrap_or(false);
if !audio_disabled && audio::pipewire_available() {
features |= FEATURE_AUDIO;
}
}
let _ = send_outbox(c, msg_hello(1, features));
}
let mut initial_msgs = Vec::with_capacity(2 + sess.ptys.len() * 2);
// Send surface-created messages BEFORE the PTY list so that
// the client's surface store is populated before `ready` is
// set — otherwise the BSP reconciliation runs with an empty
// surface list and wipes restored surface assignments.
if let Some(cs) = sess.compositor.as_ref() {
for info in cs.surfaces.values() {
// Use the latest known pixel dimensions if the stored
// width/height is still 0 (surface created before first commit).
let (w, h) = if info.width == 0 && info.height == 0 {
cs.last_pixels
.get(&info.surface_id)
.map(|lp| (lp.width as u16, lp.height as u16))
.unwrap_or((0, 0))
} else {
(info.width, info.height)
};
initial_msgs.push(msg_surface_created(
info.surface_id,
info.parent_id,
w,
h,
&info.title,
&info.app_id,
));
// Also send a resize message so the client gets the
// correct dimensions even if surface_created carried 0x0.
if w > 0 && h > 0 {
initial_msgs.push(msg_surface_resized(info.surface_id, w, h));
}
}
}
initial_msgs.push(sess.pty_list_msg());
for (&id, pty) in &sess.ptys {
let title = pty.driver.title();
if !title.is_empty() {
let title_bytes = title.as_bytes();
let mut msg = Vec::with_capacity(3 + title_bytes.len());
msg.push(S2C_TITLE);
msg.extend_from_slice(&id.to_le_bytes());
msg.extend_from_slice(title_bytes);
initial_msgs.push(msg);
}
if pty.exited {
initial_msgs.push(blit_remote::msg_exited(id, pty.exit_status));
}
}
initial_msgs.push(vec![S2C_READY]);
let tx = sess.clients.get(&client_id).map(|c| {
(
c.tx.clone(),
c.outbox_queued_frames.clone(),
c.outbox_queued_bytes.clone(),
)
});
drop(sess);
if let Some((tx, queued_frames, queued_bytes)) = tx {
for msg in initial_msgs {
if send_outbox_tracked(&tx, &queued_frames, &queued_bytes, msg).is_err() {
break;
}
}
}
}
if state.config.verbose {
eprintln!("client connected");
}
while let Some(data) = read_frame(&mut reader).await {
if data.is_empty() {
continue;
}
if data[0] == C2S_ACK {
let mut sess = state.session.lock().await;
if let Some(c) = sess.clients.get_mut(&client_id) {
c.acks_recv += 1;
record_ack(c);
} else {
continue;
}
maybe_log_pacing_metrics(&mut sess, client_id, config.verbose);
nudge_delivery(&state);
continue;
}
if data[0] == C2S_PING {
// Application-level keepalive — no-op. Its arrival is enough
// to keep the connection alive (any received data resets
// transport-level timeouts).
continue;
}
if data[0] == C2S_DISPLAY_RATE && data.len() >= 3 {
let fps = u16::from_le_bytes([data[1], data[2]]) as f32;
if fps > 0.0 {
let mut sess = state.session.lock().await;
if let Some(c) = sess.clients.get_mut(&client_id) {
c.display_fps = fps;
}
// Advertise the highest refresh rate across all clients
// to the compositor so Wayland apps render at full speed.
let max_fps = sess
.clients
.values()
.map(|c| c.display_fps)
.fold(0.0f32, f32::max);
let mhz = (max_fps * 1000.0).round() as u32;
if mhz > 0
&& let Some(cs) = &sess.compositor
{
let _ = cs
.handle
.command_tx
.send(blit_compositor::CompositorCommand::SetRefreshRate { mhz });
}
}
nudge_delivery(&state);
continue;
}
if data[0] == C2S_CLIENT_METRICS && data.len() >= 7 {
let backlog_frames = u16::from_le_bytes([data[1], data[2]]);
let ack_ahead_frames = u16::from_le_bytes([data[3], data[4]]);
let apply_ms = u16::from_le_bytes([data[5], data[6]]) as f32 * 0.1;
let mut sess = state.session.lock().await;
if let Some(c) = sess.clients.get_mut(&client_id) {
c.browser_backlog_frames = backlog_frames;
c.browser_ack_ahead_frames = ack_ahead_frames;
c.browser_apply_ms = apply_ms;
c.last_metrics_update = Instant::now();
}
nudge_delivery(&state);
continue;
}
// Server-side mouse: client sends structured mouse data, server generates
// the correct escape sequence using the terminal's current mouse mode/encoding.
if data[0] == C2S_MOUSE && data.len() >= 9 {
let pid = u16::from_le_bytes([data[1], data[2]]);
let type_ = data[3];
let button = data[4];
let col = u16::from_le_bytes([data[5], data[6]]);
let row = u16::from_le_bytes([data[7], data[8]]);
let sess = state.session.lock().await;
if let Some(pty) = sess.ptys.get(&pid) {
let (echo, icanon) = pty.lflag_cache;
if let Some(seq) = pty
.driver
.mouse_event(type_, button, col, row, echo, icanon)
&& let Some(&fd) = state.pty_fds.read().unwrap().get(&pid)
{
pty::pty_write_all(fd, &seq);
}
}
continue;
}
if data[0] == C2S_INPUT && data.len() >= 3 {
let pid = u16::from_le_bytes([data[1], data[2]]);
let mut need_nudge = false;
{
let mut sess = state.session.lock().await;
if let Some(c) = sess.clients.get_mut(&client_id)
&& update_client_scroll_state(c, pid, 0)
&& let Some(pty) = sess.ptys.get_mut(&pid)
{
pty.mark_dirty();
need_nudge = true;
}
}
if need_nudge {
nudge_delivery(&state);
}
if let Some(&fd) = state.pty_fds.read().unwrap().get(&pid) {
pty::pty_write_all(fd, &data[3..]);
}
continue;
}
if data[0] == C2S_SEARCH && data.len() >= 3 {
let request_id = u16::from_le_bytes([data[1], data[2]]);
let query = std::str::from_utf8(&data[3..]).unwrap_or("").trim();
let mut sess = state.session.lock().await;
let lead = sess.clients.get(&client_id).and_then(|c| c.lead);
let mut ranked: Vec<SearchResultRow> = if query.is_empty() {
Vec::new()
} else {
sess.ptys
.iter()
.filter_map(|(&pty_id, pty)| {
pty.driver
.search_result(query)
.map(|result| SearchResultRow {
pty_id,
score: result.score,
primary_source: result.primary_source,
matched_sources: result.matched_sources,
context: result.context,
scroll_offset: result.scroll_offset,
})
})
.collect()
};
ranked.sort_by(|a, b| {
b.score
.cmp(&a.score)
.then_with(|| (Some(b.pty_id) == lead).cmp(&(Some(a.pty_id) == lead)))
.then_with(|| a.pty_id.cmp(&b.pty_id))
});
if let Some(client) = sess.clients.get_mut(&client_id) {
let _ = send_outbox(client, build_search_results_msg(request_id, &ranked));
}
continue;
}
if data[0] == C2S_SURFACE_CAPTURE && data.len() >= 3 {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
// Extended message includes format and quality bytes.
let format = data.get(3).copied().unwrap_or(CAPTURE_FORMAT_PNG);
let quality = data.get(4).copied().unwrap_or(0);
let scale_120 = if data.len() >= 7 {
u16::from_le_bytes([data[5], data[6]])
} else {
0
};
let mut reply_msg = vec![S2C_SURFACE_CAPTURE];
reply_msg.extend_from_slice(&surface_id.to_le_bytes());
eprintln!("[capture] acquiring lock for surface {surface_id}");
let (snapshot, command_tx) = {
let sess = state.session.lock().await;
eprintln!("[capture] lock acquired");
let snap = sess
.compositor
.as_ref()
.and_then(|cs| cs.last_pixels.get(&surface_id))
.map(|lp| (lp.width, lp.height, lp.pixels.clone()));
let cmd_tx = sess
.compositor
.as_ref()
.map(|cs| cs.handle.command_tx.clone());
(snap, cmd_tx)
};
// Compositor direct capture (CPU compositing from the per-surface
// pixel cache). This is the primary path — it produces correct
// lossless results for clients that use CPU-mappable DMA-BUFs
// (Chromium/Brave) or SHM buffers.
let mut captured: Option<(u32, u32, Vec<u8>)> = None;
if let Some(ctx) = command_tx {
captured = request_surface_capture_with_timeout(
ctx,
surface_id,
scale_120,
Duration::from_secs(5),
)
.await;
}
// Fallback: last_pixels from the video pipeline. Used when
// the compositor capture returns nothing (no cached buffers).
if captured.is_none() {
captured = snapshot.and_then(|(w, h, pixels)| {
if pixels.is_dmabuf() {
return None;
}
let rgba = pixels.to_rgba(w, h);
if rgba.is_empty() {
None
} else {
Some((w, h, rgba))
}
});
}
eprintln!("[capture] acquiring client_tx lock");
let client_tx = {
let sess = state.session.lock().await;
eprintln!("[capture] client_tx lock acquired");
sess.clients.get(&client_id).map(|c| {
(
c.tx.clone(),
c.outbox_queued_frames.clone(),
c.outbox_queued_bytes.clone(),
)
})
};
if let Some((w, h, rgba_pixels)) = captured {
let image_data = encode_capture(&rgba_pixels, w, h, format, quality);
reply_msg.extend_from_slice(&w.to_le_bytes());
reply_msg.extend_from_slice(&h.to_le_bytes());
reply_msg.extend_from_slice(&image_data);
} else {
reply_msg.extend_from_slice(&0u32.to_le_bytes());
reply_msg.extend_from_slice(&0u32.to_le_bytes());
}
if let Some((client_tx, queued_frames, queued_bytes)) = client_tx {
eprintln!("[capture] sending reply: {} bytes", reply_msg.len());
match send_outbox_tracked(&client_tx, &queued_frames, &queued_bytes, reply_msg) {
Ok(()) => eprintln!("[capture] sent OK"),
Err(e) => eprintln!("[capture] send failed: {e}"),
}
} else {
eprintln!("[capture] no client_tx");
}
continue;
}
if data[0] == C2S_QUIT {
let sess = state.session.lock().await;
sess.send_to_all(&[S2C_QUIT]);
drop(sess);
state.shutdown_notify.notify_one();
break;
}
let mut sess = state.session.lock().await;
let mut need_nudge = false;
match data[0] {
C2S_SCROLL if data.len() >= 7 => {
let pid = u16::from_le_bytes([data[1], data[2]]);
let offset = u32::from_le_bytes([data[3], data[4], data[5], data[6]]) as usize;
if sess.ptys.contains_key(&pid) {
if let Some(c) = sess.clients.get_mut(&client_id) {
update_client_scroll_state(c, pid, offset);
}
if let Some(pty) = sess.ptys.get_mut(&pid) {
pty.mark_dirty();
need_nudge = true;
}
}
}
C2S_RESIZE if data.len() >= 7 => {
let entries = data[1..].chunks_exact(6);
if !entries.remainder().is_empty() {
continue;
}
let mut touched = Vec::with_capacity((data.len() - 1) / 6);
for entry in entries {
let pid = u16::from_le_bytes([entry[0], entry[1]]);
if !sess.ptys.contains_key(&pid) {
continue;
}
let rows = u16::from_le_bytes([entry[2], entry[3]]);
let cols = u16::from_le_bytes([entry[4], entry[5]]);
if let Some(c) = sess.clients.get_mut(&client_id) {
if is_unset_view_size(rows, cols) {
if c.view_sizes.remove(&pid).is_some() {
touched.push(pid);
}
} else if rows == 0 || cols == 0 {
continue;
} else {
c.view_sizes.insert(pid, (rows, cols));
touched.push(pid);
}
}
}
if sess.resize_ptys_to_mediated_sizes(touched) {
need_nudge = true;
}
}
C2S_CREATE => {
// Format: [opcode][rows:2][cols:2][tag_len:2][tag:N][command...]
let (rows, cols) = if data.len() >= 5 {
(
u16::from_le_bytes([data[1], data[2]]),
u16::from_le_bytes([data[3], data[4]]),
)
} else {
(24, 80)
};
let tag_len = if data.len() >= 7 {
u16::from_le_bytes([data[5], data[6]]) as usize
} else {
0
};
let tag = if data.len() >= 7 + tag_len {
std::str::from_utf8(&data[7..7 + tag_len]).unwrap_or_default()
} else {
""
};
let cmd_start = 7 + tag_len;
let dir: Option<String> = None;
let create_payload = data
.get(cmd_start..)
.and_then(|bytes| std::str::from_utf8(bytes).ok());
let command = create_payload
.filter(|payload| !payload.contains('\0'))
.map(str::trim)
.filter(|payload| !payload.is_empty());
let argv: Option<Vec<&str>> = create_payload
.filter(|payload| payload.contains('\0'))
.map(|payload| {
payload
.split('\0')
.filter(|arg| !arg.is_empty())
.collect::<Vec<_>>()
})
.filter(|args| !args.is_empty());
let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
continue;
};
let socket_name = sess
.ensure_compositor(
config.verbose,
notify_for_compositor.clone(),
&config.vaapi_device,
)
.to_string();
#[cfg(target_os = "linux")]
let pulse_server = sess.pulse_server_path();
#[cfg(not(target_os = "linux"))]
let pulse_server: Option<String> = None;
#[cfg(target_os = "linux")]
let pipewire_remote = sess.pipewire_remote_path();
#[cfg(not(target_os = "linux"))]
let pipewire_remote: Option<String> = None;
if let Some(pty) = pty::spawn_pty(
&config.shell,
&config.shell_flags,
rows,
cols,
id,
tag,
command,
argv.as_deref(),
dir.as_deref(),
config.scrollback,
state.clone(),
Some(&socket_name),
pulse_server.as_deref(),
pipewire_remote.as_deref(),
) {
let mut msg = Vec::with_capacity(3 + pty.tag.len());
msg.push(S2C_CREATED);
msg.extend_from_slice(&id.to_le_bytes());
msg.extend_from_slice(pty.tag.as_bytes());
sess.ptys.insert(id, pty);
if let Some(c) = sess.clients.get_mut(&client_id) {
c.lead = Some(id);
c.view_sizes.insert(id, (rows, cols));
subscribe_client_to(c, id);
reset_inflight(c);
}
sess.send_to_all(&msg);
need_nudge = true;
}
}
C2S_CREATE_N => {
// Format: [opcode][nonce:2][rows:2][cols:2][tag_len:2][tag:N][command...]
let nonce = if data.len() >= 3 {
u16::from_le_bytes([data[1], data[2]])
} else {
0
};
let (rows, cols) = if data.len() >= 7 {
(
u16::from_le_bytes([data[3], data[4]]),
u16::from_le_bytes([data[5], data[6]]),
)
} else {
(24, 80)
};
let tag_len = if data.len() >= 9 {
u16::from_le_bytes([data[7], data[8]]) as usize
} else {
0
};
let tag = if data.len() >= 9 + tag_len {
std::str::from_utf8(&data[9..9 + tag_len]).unwrap_or_default()
} else {
""
};
let cmd_start = 9 + tag_len;
let dir: Option<String> = None;
let create_payload = data
.get(cmd_start..)
.and_then(|bytes| std::str::from_utf8(bytes).ok());
let command = create_payload
.filter(|payload| !payload.contains('\0'))
.map(str::trim)
.filter(|payload| !payload.is_empty());
let argv: Option<Vec<&str>> = create_payload
.filter(|payload| payload.contains('\0'))
.map(|payload| {
payload
.split('\0')
.filter(|arg| !arg.is_empty())
.collect::<Vec<_>>()
})
.filter(|args| !args.is_empty());
let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
continue;
};
let socket_name = sess
.ensure_compositor(
config.verbose,
notify_for_compositor.clone(),
&config.vaapi_device,
)
.to_string();
#[cfg(target_os = "linux")]
let pulse_server = sess.pulse_server_path();
#[cfg(not(target_os = "linux"))]
let pulse_server: Option<String> = None;
#[cfg(target_os = "linux")]
let pipewire_remote = sess.pipewire_remote_path();
#[cfg(not(target_os = "linux"))]
let pipewire_remote: Option<String> = None;
if let Some(pty) = pty::spawn_pty(
&config.shell,
&config.shell_flags,
rows,
cols,
id,
tag,
command,
argv.as_deref(),
dir.as_deref(),
config.scrollback,
state.clone(),
Some(&socket_name),
pulse_server.as_deref(),
pipewire_remote.as_deref(),
) {
let tag_bytes = pty.tag.as_bytes();
let mut nonce_msg = Vec::with_capacity(5 + tag_bytes.len());
nonce_msg.push(S2C_CREATED_N);
nonce_msg.extend_from_slice(&nonce.to_le_bytes());
nonce_msg.extend_from_slice(&id.to_le_bytes());
nonce_msg.extend_from_slice(tag_bytes);
let mut broadcast_msg = Vec::with_capacity(3 + tag_bytes.len());
broadcast_msg.push(S2C_CREATED);
broadcast_msg.extend_from_slice(&id.to_le_bytes());
broadcast_msg.extend_from_slice(tag_bytes);
sess.ptys.insert(id, pty);
if let Some(c) = sess.clients.get_mut(&client_id) {
c.lead = Some(id);
c.view_sizes.insert(id, (rows, cols));
subscribe_client_to(c, id);
reset_inflight(c);
let _ = send_outbox(c, nonce_msg);
}
for (&cid, c) in sess.clients.iter() {
if cid != client_id {
let _ = send_outbox(c, broadcast_msg.clone());
}
}
need_nudge = true;
}
}
C2S_CREATE_AT => {
// Format: [opcode][rows:2][cols:2][tag_len:2][tag:N][src_pty_id:2]
let (rows, cols) = if data.len() >= 5 {
(
u16::from_le_bytes([data[1], data[2]]),
u16::from_le_bytes([data[3], data[4]]),
)
} else {
(24, 80)
};
let tag_len = if data.len() >= 7 {
u16::from_le_bytes([data[5], data[6]]) as usize
} else {
0
};
let tag = if data.len() >= 7 + tag_len {
std::str::from_utf8(&data[7..7 + tag_len]).unwrap_or_default()
} else {
""
};
let src_start = 7 + tag_len;
let dir = if data.len() >= src_start + 2 {
let src_id = u16::from_le_bytes([data[src_start], data[src_start + 1]]);
sess.ptys.get(&src_id).and_then(|p| pty::pty_cwd(&p.handle))
} else {
None
};
let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
continue;
};
let socket_name = sess
.ensure_compositor(
config.verbose,
notify_for_compositor.clone(),
&config.vaapi_device,
)
.to_string();
#[cfg(target_os = "linux")]
let pulse_server = sess.pulse_server_path();
#[cfg(not(target_os = "linux"))]
let pulse_server: Option<String> = None;
#[cfg(target_os = "linux")]
let pipewire_remote = sess.pipewire_remote_path();
#[cfg(not(target_os = "linux"))]
let pipewire_remote: Option<String> = None;
if let Some(pty) = pty::spawn_pty(
&config.shell,
&config.shell_flags,
rows,
cols,
id,
tag,
None,
None,
dir.as_deref(),
config.scrollback,
state.clone(),
Some(&socket_name),
pulse_server.as_deref(),
pipewire_remote.as_deref(),
) {
let mut msg = Vec::with_capacity(3 + pty.tag.len());
msg.push(S2C_CREATED);
msg.extend_from_slice(&id.to_le_bytes());
msg.extend_from_slice(pty.tag.as_bytes());
sess.ptys.insert(id, pty);
if let Some(c) = sess.clients.get_mut(&client_id) {
c.lead = Some(id);
c.view_sizes.insert(id, (rows, cols));
subscribe_client_to(c, id);
reset_inflight(c);
}
sess.send_to_all(&msg);
need_nudge = true;
}
}
C2S_CREATE2 => {
if data.len() < 10 {
continue;
}
let nonce = u16::from_le_bytes([data[1], data[2]]);
let rows = u16::from_le_bytes([data[3], data[4]]);
let cols = u16::from_le_bytes([data[5], data[6]]);
let features = data[7];
let tag_len = u16::from_le_bytes([data[8], data[9]]) as usize;
let tag = if data.len() >= 10 + tag_len {
std::str::from_utf8(&data[10..10 + tag_len]).unwrap_or_default()
} else {
""
};
let mut cursor = 10 + tag_len;
let dir = if features & CREATE2_HAS_SRC_PTY != 0 && data.len() >= cursor + 2 {
let src_id = u16::from_le_bytes([data[cursor], data[cursor + 1]]);
cursor += 2;
sess.ptys.get(&src_id).and_then(|p| pty::pty_cwd(&p.handle))
} else {
None
};
let create_payload = if features & CREATE2_HAS_COMMAND != 0 {
data.get(cursor..).and_then(|b| std::str::from_utf8(b).ok())
} else {
None
};
let command = create_payload
.filter(|p| !p.contains('\0'))
.map(str::trim)
.filter(|p| !p.is_empty());
let argv: Option<Vec<&str>> = create_payload
.filter(|p| p.contains('\0'))
.map(|p| p.split('\0').filter(|a| !a.is_empty()).collect::<Vec<_>>())
.filter(|a| !a.is_empty());
let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
continue;
};
let socket_name = sess
.ensure_compositor(
config.verbose,
notify_for_compositor.clone(),
&config.vaapi_device,
)
.to_string();
#[cfg(target_os = "linux")]
let pulse_server = sess.pulse_server_path();
#[cfg(not(target_os = "linux"))]
let pulse_server: Option<String> = None;
#[cfg(target_os = "linux")]
let pipewire_remote = sess.pipewire_remote_path();
#[cfg(not(target_os = "linux"))]
let pipewire_remote: Option<String> = None;
if let Some(pty) = pty::spawn_pty(
&config.shell,
&config.shell_flags,
rows,
cols,
id,
tag,
command,
argv.as_deref(),
dir.as_deref(),
config.scrollback,
state.clone(),
Some(&socket_name),
pulse_server.as_deref(),
pipewire_remote.as_deref(),
) {
let tag_bytes = pty.tag.as_bytes();
let mut nonce_msg = Vec::with_capacity(5 + tag_bytes.len());
nonce_msg.push(S2C_CREATED_N);
nonce_msg.extend_from_slice(&nonce.to_le_bytes());
nonce_msg.extend_from_slice(&id.to_le_bytes());
nonce_msg.extend_from_slice(tag_bytes);
let mut broadcast_msg = Vec::with_capacity(3 + tag_bytes.len());
broadcast_msg.push(S2C_CREATED);
broadcast_msg.extend_from_slice(&id.to_le_bytes());
broadcast_msg.extend_from_slice(tag_bytes);
sess.ptys.insert(id, pty);
if let Some(c) = sess.clients.get_mut(&client_id) {
c.lead = Some(id);
c.view_sizes.insert(id, (rows, cols));
subscribe_client_to(c, id);
reset_inflight(c);
let _ = send_outbox(c, nonce_msg);
}
for (&cid, c) in sess.clients.iter() {
if cid != client_id {
let _ = send_outbox(c, broadcast_msg.clone());
}
}
need_nudge = true;
}
}
C2S_SURFACE_INPUT if data.len() >= 8 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
let keycode = u32::from_le_bytes([data[3], data[4], data[5], data[6]]);
let pressed = data[7] != 0;
if let Some(client) = sess.clients.get_mut(&client_id) {
if pressed {
client.pressed_surface_keys.insert(keycode);
} else {
client.pressed_surface_keys.remove(&keycode);
}
}
if let Some(cs) = sess.compositor.as_mut() {
let _ = cs.handle.command_tx.send(CompositorCommand::KeyInput {
surface_id,
keycode,
pressed,
});
cs.handle.wake();
state.delivery_notify.notify_one();
}
}
C2S_SURFACE_TEXT if data.len() >= 3 => {
let _surface_id = u16::from_le_bytes([data[1], data[2]]);
if let Ok(text) = std::str::from_utf8(&data[3..])
&& let Some(cs) = sess.compositor.as_mut()
{
let _ = cs.handle.command_tx.send(CompositorCommand::TextInput {
text: text.to_string(),
});
cs.handle.wake();
state.delivery_notify.notify_one();
}
}
C2S_SURFACE_POINTER if data.len() >= 9 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
let ptype = data[3];
let button = data[4];
let x = u16::from_le_bytes([data[5], data[6]]) as f64;
let y = u16::from_le_bytes([data[7], data[8]]) as f64;
if let Some(cs) = sess.compositor.as_mut() {
match ptype {
0 | 1 => {
let _ = cs.handle.command_tx.send(CompositorCommand::PointerMotion {
surface_id,
x,
y,
});
let _ = cs.handle.command_tx.send(CompositorCommand::PointerButton {
surface_id,
button: match button {
1 => 0x112,
2 => 0x111,
_ => 0x110,
},
pressed: ptype == 0,
});
}
2 => {
let _ = cs.handle.command_tx.send(CompositorCommand::PointerMotion {
surface_id,
x,
y,
});
}
_ => {}
}
cs.handle.wake();
}
state.delivery_notify.notify_one();
}
C2S_SURFACE_POINTER_AXIS if data.len() >= 8 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
let axis = data[3];
let value_x100 = i32::from_le_bytes([data[4], data[5], data[6], data[7]]);
let value = value_x100 as f64 / 100.0;
if let Some(cs) = sess.compositor.as_mut() {
let _ = cs.handle.command_tx.send(CompositorCommand::PointerAxis {
surface_id,
axis,
value,
});
cs.handle.wake();
}
state.delivery_notify.notify_one();
}
C2S_SURFACE_RESIZE if data.len() >= 9 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
let width = u16::from_le_bytes([data[3], data[4]]);
let height = u16::from_le_bytes([data[5], data[6]]);
// Scale in 1/120th units (Wayland convention): 240 = 2×.
let scale_120 = u16::from_le_bytes([data[7], data[8]]);
if state.config.verbose {
eprintln!(
"C2S_SURFACE_RESIZE: cid={client_id} sid={surface_id} {width}x{height} scale={scale_120}"
);
}
if let Some(c) = sess.clients.get_mut(&client_id) {
if is_unset_view_size(width, height) {
c.surface_view_sizes.remove(&surface_id);
} else if width > 0 && height > 0 {
c.surface_view_sizes
.insert(surface_id, (width, height, scale_120));
}
// Clear latched nal_data=None streak for this
// surface so the encoder can be recreated. The
// streak is designed to stop infinite recreation
// loops (GBM fd leak), not to permanently black out
// a surface across a client-driven resize.
//
// Also wake the pacing gate: reset the burst window
// and clear next_send_at so the first frame after
// resize bypasses time-based pacing and flows at
// wire speed. Without this, the client waits up to
// one send interval (~1/fps) after the encoder is
// recreated before seeing the first new frame.
if let Some(s) = c.surface_subs.get_mut(&surface_id) {
s.nal_none_streak = 0;
s.nal_none_latched_at = None;
s.burst_remaining = SURFACE_BURST_FRAMES;
s.next_send_at = None;
}
c.surface_needs_keyframe = true;
}
sess.resize_surfaces_to_mediated_sizes(
std::iter::once(surface_id),
&state.config.surface_encoders,
);
}
C2S_SURFACE_FOCUS if data.len() >= 3 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
if state.config.verbose {
eprintln!("C2S_SURFACE_FOCUS: cid={client_id} sid={surface_id}");
}
if let Some(cs) = sess.compositor.as_ref() {
let _ = cs
.handle
.command_tx
.send(CompositorCommand::SurfaceFocus { surface_id });
}
}
C2S_SURFACE_CLOSE if data.len() >= 3 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
if let Some(cs) = sess.compositor.as_ref() {
let _ = cs
.handle
.command_tx
.send(CompositorCommand::SurfaceClose { surface_id });
cs.handle.wake();
}
}
C2S_SURFACE_SUBSCRIBE if data.len() >= 3 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
// Extended fields (backward-compatible: absent = 0 = defaults).
let codec_support = if data.len() >= 4 { data[3] } else { 0 };
let quality_wire = if data.len() >= 5 { data[4] } else { 0 };
if state.config.verbose {
eprintln!(
"C2S_SURFACE_SUBSCRIBE: cid={client_id} surface={surface_id} codec={codec_support:#04x} quality={quality_wire}"
);
}
let mut destroy_vulkan_enc_sid = None;
let mut first_subscribe = false;
if let Some(c) = sess.clients.get_mut(&client_id) {
let was_subscribed = !c.surface_subscriptions.insert(surface_id);
let new_quality = SurfaceQuality::from_wire(quality_wire);
let state = c.surface_subs.entry(surface_id).or_default();
let old_codec = state.codec_override;
let old_quality = state.quality_override;
// A no-op resubscribe (same codec/quality, already
// subscribed) should not disturb the steady encode
// stream — resetting needs_keyframe/burst on every
// repeated subscribe makes keyframes churn and skews
// pacing.
let meaningful_change =
!was_subscribed || codec_support != old_codec || new_quality != old_quality;
state.codec_override = codec_support;
state.quality_override = new_quality;
let task_in_flight = state.encode_in_flight || state.creation_in_flight;
if meaningful_change {
// Reset burst window so the first frames after a
// (re)subscribe bypass time-based pacing and flow
// at wire speed. Clear the nal_data=None streak
// too: a fresh subscription is a valid signal to
// retry a previously-latched encoder.
state.burst_remaining = SURFACE_BURST_FRAMES;
state.nal_none_streak = 0;
state.nal_none_latched_at = None;
}
// Force encoder recreation when preferences change on
// resubscribe. If an encode OR creation is in flight,
// flag the completion handler to discard its encoder
// instead of installing the stale one.
if was_subscribed && (codec_support != old_codec || new_quality != old_quality)
{
state.encoder = None;
if task_in_flight {
state.encoder_invalidated = true;
}
}
if meaningful_change {
c.surface_needs_keyframe = true;
}
first_subscribe = !was_subscribed;
if was_subscribed
&& (codec_support != old_codec || new_quality != old_quality)
&& c.vulkan_video_surfaces.remove(&surface_id).is_some()
{
destroy_vulkan_enc_sid = Some(surface_id);
}
}
if let Some(sid) = destroy_vulkan_enc_sid
&& let Some(cs) = sess.compositor.as_ref()
{
let _ = cs.handle.command_tx.send(
blit_compositor::CompositorCommand::DestroyVulkanEncoder {
surface_id: sid as u32,
},
);
cs.handle.wake();
}
if first_subscribe {
sess.resize_surfaces_to_mediated_sizes(
std::iter::once(surface_id),
&state.config.surface_encoders,
);
}
state.delivery_notify.notify_one();
}
C2S_SURFACE_UNSUBSCRIBE if data.len() >= 3 => {
let surface_id = u16::from_le_bytes([data[1], data[2]]);
let mut removed_vulkan = false;
if let Some(c) = sess.clients.get_mut(&client_id) {
c.surface_subscriptions.remove(&surface_id);
c.surface_subs.remove(&surface_id);
removed_vulkan = c.vulkan_video_surfaces.remove(&surface_id).is_some();
c.surface_view_sizes.remove(&surface_id);
}
// Destroy Vulkan Video encoder if no remaining client needs it.
if removed_vulkan {
let still_needed = sess
.clients
.values()
.any(|other| other.vulkan_video_surfaces.contains_key(&surface_id));
if !still_needed && let Some(cs) = sess.compositor.as_ref() {
let _ = cs.handle.command_tx.send(
blit_compositor::CompositorCommand::DestroyVulkanEncoder {
surface_id: surface_id as u32,
},
);
cs.handle.wake();
}
}
sess.resize_surfaces_to_mediated_sizes(
std::iter::once(surface_id),
&state.config.surface_encoders,
);
}
#[cfg(target_os = "linux")]
C2S_AUDIO_SUBSCRIBE if data.len() >= 3 => {
let bitrate_kbps = u16::from_le_bytes([data[1], data[2]]);
let audio_tx = sess.clients.get(&client_id).map(|c| c.audio_tx.clone());
if let Some(c) = sess.clients.get_mut(&client_id) {
c.audio_subscribed = true;
c.audio_bitrate_kbps = bitrate_kbps;
if state.config.verbose {
eprintln!(
"C2S_AUDIO_SUBSCRIBE: cid={client_id} bitrate_kbps={bitrate_kbps}"
);
}
}
// Register with the audio broadcast — atomically enqueues
// catch-up frames and registers for live frames from the
// fan-out task. Succeeds even if the pipeline itself is
// currently down (frames resume once it's respawned).
if let (Some(cs), Some(tx)) = (sess.compositor.as_ref(), audio_tx) {
cs.audio_broadcast.subscribe(client_id, tx);
}
// Recompute the effective audio bitrate across all
// subscribed clients (use the max requested bitrate).
if let Some(cs) = sess.compositor.as_ref()
&& let Some(ref ap) = cs.audio_pipeline
{
let max_kbps = sess
.clients
.values()
.filter(|c| c.audio_subscribed)
.map(|c| c.audio_bitrate_kbps)
.max()
.unwrap_or(0);
let bitrate = if max_kbps > 0 {
max_kbps as i32 * 1000
} else {
audio::DEFAULT_BITRATE
};
ap.set_bitrate(bitrate);
}
state.delivery_notify.notify_one();
}
#[cfg(target_os = "linux")]
C2S_AUDIO_UNSUBSCRIBE if !data.is_empty() => {
if let Some(c) = sess.clients.get_mut(&client_id) {
c.audio_subscribed = false;
c.audio_bitrate_kbps = 0;
if state.config.verbose {
eprintln!("C2S_AUDIO_UNSUBSCRIBE: cid={client_id}");
}
}
if let Some(cs) = sess.compositor.as_ref() {
cs.audio_broadcast.unsubscribe(client_id);
}
// Recompute effective bitrate after unsubscribe.
if let Some(cs) = sess.compositor.as_ref()
&& let Some(ref ap) = cs.audio_pipeline
{
let max_kbps = sess
.clients
.values()
.filter(|c| c.audio_subscribed)
.map(|c| c.audio_bitrate_kbps)
.max()
.unwrap_or(0);
let bitrate = if max_kbps > 0 {
max_kbps as i32 * 1000
} else {
audio::DEFAULT_BITRATE
};
ap.set_bitrate(bitrate);
}
}
C2S_SURFACE_ACK if data.len() >= 3 => {
// Surface ACKs feed shared RTT / delivery_bps / goodput_bps
// from a separate inflight queue so they don't corrupt
// terminal frame-size averages or probe_frames.
if let Some(c) = sess.clients.get_mut(&client_id) {
c.acks_recv += 1;
record_surface_ack(c);
}
state.delivery_notify.notify_one();
}
C2S_CLIENT_FEATURES if data.len() >= 2 => {
// Byte 0: codec_support bitmask. Future bytes are ignored
// if unknown, defaulting to 0 when absent.
let codec_support = data[1];
if let Some(c) = sess.clients.get_mut(&client_id) {
c.surface_codec_support = codec_support;
}
}
C2S_CLIPBOARD_SET if data.len() >= 5 => {
let mime_len = u16::from_le_bytes([data[1], data[2]]) as usize;
if data.len() >= 3 + mime_len + 4 {
let mime = std::str::from_utf8(&data[3..3 + mime_len])
.unwrap_or("text/plain")
.to_string();
let data_len = u32::from_le_bytes([
data[3 + mime_len],
data[4 + mime_len],
data[5 + mime_len],
data[6 + mime_len],
]) as usize;
let payload_start = 7 + mime_len;
if data.len() >= payload_start + data_len {
let payload = data[payload_start..payload_start + data_len].to_vec();
if let Some(cs) = sess.compositor.as_ref() {
let _ = cs
.handle
.command_tx
.send(CompositorCommand::ClipboardOffer {
mime_type: mime,
data: payload,
});
}
}
}
}
C2S_CLIPBOARD_LIST if !data.is_empty() => {
if let Some(cs) = sess.compositor.as_ref() {
let command_tx = cs.handle.command_tx.clone();
let client_tx = sess.clients.get(&client_id).map(|c| {
(
c.tx.clone(),
c.outbox_queued_frames.clone(),
c.outbox_queued_bytes.clone(),
)
});
if let Some((client_tx, queued_frames, queued_bytes)) = client_tx {
tokio::task::spawn_blocking(move || {
let (tx, rx) = std::sync::mpsc::sync_channel(1);
if command_tx
.send(CompositorCommand::ClipboardListMimes { reply: tx })
.is_ok()
&& let Ok(mimes) = rx.recv_timeout(Duration::from_secs(2))
{
let _ = send_outbox_tracked(
&client_tx,
&queued_frames,
&queued_bytes,
msg_s2c_clipboard_list(&mimes),
);
}
});
}
} else {
// No compositor — respond with empty list.
if let Some(c) = sess.clients.get(&client_id) {
let _ = send_outbox(c, msg_s2c_clipboard_list(&[]));
}
}
}
C2S_CLIPBOARD_GET if data.len() >= 3 => {
let mime_len = u16::from_le_bytes([data[1], data[2]]) as usize;
if data.len() >= 3 + mime_len {
let mime = std::str::from_utf8(&data[3..3 + mime_len])
.unwrap_or("text/plain")
.to_string();
if let Some(cs) = sess.compositor.as_ref() {
let command_tx = cs.handle.command_tx.clone();
let client_tx = sess.clients.get(&client_id).map(|c| {
(
c.tx.clone(),
c.outbox_queued_frames.clone(),
c.outbox_queued_bytes.clone(),
)
});
if let Some((client_tx, queued_frames, queued_bytes)) = client_tx {
tokio::task::spawn_blocking(move || {
let (tx, rx) = std::sync::mpsc::sync_channel(1);
if command_tx
.send(CompositorCommand::ClipboardGet {
mime_type: mime.clone(),
reply: tx,
})
.is_ok()
&& let Ok(content) = rx.recv_timeout(Duration::from_secs(2))
{
let data = content.unwrap_or_default();
let _ = send_outbox_tracked(
&client_tx,
&queued_frames,
&queued_bytes,
msg_s2c_clipboard_content(&mime, &data),
);
}
});
}
} else {
// No compositor — respond with empty clipboard.
if let Some(c) = sess.clients.get(&client_id) {
let _ = send_outbox(c, msg_s2c_clipboard_content(&mime, &[]));
}
}
}
}
C2S_SURFACE_LIST if !data.is_empty() => {
let msg = sess.surface_list_msg();
if let Some(c) = sess.clients.get(&client_id) {
let _ = send_outbox(c, msg);
}
}
C2S_FOCUS if data.len() >= 3 => {
let pid = u16::from_le_bytes([data[1], data[2]]);
if sess.ptys.contains_key(&pid) {
let old_pid = sess.clients.get(&client_id).and_then(|c| c.lead);
if let Some(c) = sess.clients.get_mut(&client_id) {
c.lead = Some(pid);
subscribe_client_to(c, pid);
if old_pid == Some(pid) {
update_client_scroll_state(c, pid, 0);
} else {
reset_inflight(c);
}
}
if let Some(pty) = sess.ptys.get_mut(&pid) {
pty.mark_dirty();
need_nudge = true;
}
}
}
C2S_SUBSCRIBE if data.len() >= 3 => {
let pid = u16::from_le_bytes([data[1], data[2]]);
if sess.ptys.contains_key(&pid) {
if let Some(c) = sess.clients.get_mut(&client_id) {
subscribe_client_to(c, pid);
}
if let Some(pty) = sess.ptys.get_mut(&pid) {
pty.mark_dirty();
}
need_nudge = true;
}
}
C2S_UNSUBSCRIBE if data.len() >= 3 => {
let pid = u16::from_le_bytes([data[1], data[2]]);
if sess.ptys.contains_key(&pid) {
let mut touched = Vec::new();
if let Some(c) = sess.clients.get_mut(&client_id) {
if unsubscribe_client_from(c, pid) {
touched.push(pid);
}
reset_inflight(c);
}
if sess.resize_ptys_to_mediated_sizes(touched) {
need_nudge = true;
}
}
}
C2S_RESTART if data.len() >= 3 => {
let pid = u16::from_le_bytes([data[1], data[2]]);
let restart_info = sess
.ptys
.get(&pid)
.filter(|p| p.exited)
.map(|p| (p.driver.size(), p.command.clone(), p.tag.clone()));
if let Some(((rows, cols), command, tag)) = restart_info {
let wayland_display = sess
.compositor
.as_ref()
.map(|cs| cs.handle.socket_name.clone());
#[cfg(target_os = "linux")]
let pulse_server = sess.pulse_server_path();
#[cfg(not(target_os = "linux"))]
let pulse_server: Option<String> = None;
#[cfg(target_os = "linux")]
let pipewire_remote = sess.pipewire_remote_path();
#[cfg(not(target_os = "linux"))]
let pipewire_remote: Option<String> = None;
if let Some((new_handle, reader, byte_rx)) = pty::respawn_child(
&state.config.shell,
&state.config.shell_flags,
rows,
cols,
pid,
command.as_deref(),
state.clone(),
wayland_display.as_deref(),
pulse_server.as_deref(),
pipewire_remote.as_deref(),
) {
let Some(pty) = sess.ptys.get_mut(&pid) else {
break;
};
pty.handle = new_handle;
pty.reader_handle = reader;
pty.byte_rx = byte_rx;
pty.driver.reset_modes();
pty.exited = false;
pty.exit_status = blit_remote::EXIT_STATUS_UNKNOWN;
pty.lflag_cache = pty::pty_lflag(&pty.handle);
pty.lflag_last = Instant::now();
pty.mark_dirty();
if let Some(c) = sess.clients.get_mut(&client_id) {
c.lead = Some(pid);
subscribe_client_to(c, pid);
update_client_scroll_state(c, pid, 0);
reset_inflight(c);
}
let mut msg = Vec::with_capacity(3 + tag.len());
msg.push(S2C_CREATED);
msg.extend_from_slice(&pid.to_le_bytes());
msg.extend_from_slice(tag.as_bytes());
sess.send_to_all(&msg);
need_nudge = true;
}
}
}
C2S_READ if data.len() >= 13 => {
let nonce = u16::from_le_bytes([data[1], data[2]]);
let pid = u16::from_le_bytes([data[3], data[4]]);
let req_offset = u32::from_le_bytes([data[5], data[6], data[7], data[8]]) as usize;
let req_limit =
u32::from_le_bytes([data[9], data[10], data[11], data[12]]) as usize;
let flags = data.get(13).copied().unwrap_or(0);
let ansi = flags & READ_ANSI != 0;
let tail = flags & READ_TAIL != 0;
if let Some(pty) = sess.ptys.get_mut(&pid) {
let (rows, _cols) = pty.driver.size();
let viewport = take_snapshot(pty);
let scrollback_lines = viewport.scrollback_lines() as usize;
let total_lines = scrollback_lines + rows as usize;
let extract = |f: &FrameState| -> String {
if ansi {
f.get_ansi_text()
} else {
f.get_all_text()
}
};
let mut all_lines: Vec<String> =
Vec::with_capacity(scrollback_lines + rows as usize);
let mut scroll_offset = scrollback_lines;
while scroll_offset > 0 {
let frame = pty.driver.scrollback_frame(scroll_offset);
let page = extract(&frame);
let page_lines: Vec<&str> = page.lines().collect();
let take = if scroll_offset < rows as usize {
scroll_offset.min(page_lines.len())
} else {
page_lines.len()
};
for line in &page_lines[..take] {
all_lines.push(line.to_string());
}
if scroll_offset <= rows as usize {
break;
}
scroll_offset = scroll_offset.saturating_sub(rows as usize);
}
for line in extract(&viewport).lines() {
all_lines.push(line.to_string());
}
let (start, end) = if tail {
let end = all_lines.len().saturating_sub(req_offset);
let start = if req_limit == 0 {
0
} else {
end.saturating_sub(req_limit)
};
(start, end)
} else {
let start = req_offset.min(all_lines.len());
let end = if req_limit == 0 {
all_lines.len()
} else {
(start + req_limit).min(all_lines.len())
};
(start, end)
};
let text = all_lines[start..end].join("\n");
let mut msg = Vec::with_capacity(13 + text.len());
msg.push(S2C_TEXT);
msg.extend_from_slice(&nonce.to_le_bytes());
msg.extend_from_slice(&pid.to_le_bytes());
msg.extend_from_slice(&(total_lines as u32).to_le_bytes());
msg.extend_from_slice(&(start as u32).to_le_bytes());
msg.extend_from_slice(text.as_bytes());
if let Some(client) = sess.clients.get(&client_id) {
let _ = send_outbox(client, msg);
}
}
}
C2S_COPY_RANGE if data.len() >= 18 => {
let nonce = u16::from_le_bytes([data[1], data[2]]);
let pid = u16::from_le_bytes([data[3], data[4]]);
let start_tail = u32::from_le_bytes([data[5], data[6], data[7], data[8]]);
let start_col = u16::from_le_bytes([data[9], data[10]]);
let end_tail = u32::from_le_bytes([data[11], data[12], data[13], data[14]]);
let end_col = u16::from_le_bytes([data[15], data[16]]);
if let Some(pty) = sess.ptys.get(&pid) {
let text = pty
.driver
.get_text_range(start_tail, start_col, end_tail, end_col);
let total_lines = pty.driver.total_lines();
let mut msg = Vec::with_capacity(13 + text.len());
msg.push(S2C_TEXT);
msg.extend_from_slice(&nonce.to_le_bytes());
msg.extend_from_slice(&pid.to_le_bytes());
msg.extend_from_slice(&total_lines.to_le_bytes());
msg.extend_from_slice(&start_tail.to_le_bytes());
msg.extend_from_slice(text.as_bytes());
if let Some(client) = sess.clients.get(&client_id) {
let _ = send_outbox(client, msg);
}
}
}
C2S_KILL if data.len() >= 7 => {
let pid = u16::from_le_bytes([data[1], data[2]]);
let signal = i32::from_le_bytes([data[3], data[4], data[5], data[6]]);
if let Some(pty) = sess.ptys.get(&pid)
&& !pty.exited
{
pty::kill_pty(&pty.handle, signal);
}
}
C2S_CLOSE if data.len() >= 3 => {
let pid = u16::from_le_bytes([data[1], data[2]]);
if let Some(pty) = sess.ptys.remove(&pid) {
if !pty.exited {
state.pty_fds.write().unwrap().remove(&pid);
drop(pty.reader_handle);
pty::close_pty(&pty.handle);
}
for client in sess.clients.values_mut() {
unsubscribe_client_from(client, pid);
}
let mut msg = vec![S2C_CLOSED];
msg.extend_from_slice(&pid.to_le_bytes());
sess.send_to_all(&msg);
}
}
_ => {}
}
drop(sess);
if need_nudge {
nudge_delivery(&state);
}
}
{
let mut sess = state.session.lock().await;
let mut need_nudge = false;
// Drop any audio subscription before removing the client so the
// fan-out task doesn't hold a dead tx for the full mpsc-buffered
// lifetime.
#[cfg(target_os = "linux")]
if let Some(cs) = sess.compositor.as_ref() {
cs.audio_broadcast.unsubscribe(client_id);
}
let client = sess.clients.remove(&client_id);
let affected_ptys = client
.as_ref()
.map(|c| c.view_sizes.keys().copied().collect::<Vec<_>>())
.unwrap_or_default();
let affected_surfaces = client
.as_ref()
.map(|c| c.surface_view_sizes.keys().copied().collect::<Vec<_>>())
.unwrap_or_default();
if sess.resize_ptys_to_mediated_sizes(affected_ptys) {
need_nudge = true;
}
sess.resize_surfaces_to_mediated_sizes(affected_surfaces, &state.config.surface_encoders);
// Release any keys this client was holding when it disconnected.
// Without this, modifier keys (Shift, Ctrl, etc.) stay stuck and
// regular keys auto-repeat forever in the compositor.
if let Some(ref client) = client
&& !client.pressed_surface_keys.is_empty()
&& let Some(cs) = sess.compositor.as_mut()
{
let keycodes: Vec<u32> = client.pressed_surface_keys.iter().copied().collect();
let _ = cs
.handle
.command_tx
.send(CompositorCommand::ReleaseKeys { keycodes });
cs.handle.wake();
}
// Destroy Vulkan Video encoders for surfaces that no remaining
// client needs.
if let Some(ref client) = client
&& !client.vulkan_video_surfaces.is_empty()
&& let Some(cs) = sess.compositor.as_ref()
{
for &sid in client.vulkan_video_surfaces.keys() {
let still_needed = sess
.clients
.values()
.any(|c| c.vulkan_video_surfaces.contains_key(&sid));
if !still_needed {
let _ = cs
.handle
.command_tx
.send(CompositorCommand::DestroyVulkanEncoder {
surface_id: sid as u32,
});
}
}
cs.handle.wake();
}
drop(sess);
if need_nudge {
nudge_delivery(&state);
}
}
sender.abort();
if state.config.verbose {
eprintln!("client disconnected");
}
}
#[cfg(test)]
mod tests {
use super::*;
fn test_client_with_capacity(
_capacity: usize,
) -> (ClientState, mpsc::UnboundedReceiver<Vec<u8>>) {
let (tx, rx) = mpsc::unbounded_channel();
let (audio_tx, _audio_rx) = mpsc::unbounded_channel();
let client = ClientState {
tx,
outbox_queued_frames: Arc::new(AtomicUsize::new(0)),
outbox_queued_bytes: Arc::new(AtomicUsize::new(0)),
audio_tx,
lead: None,
subscriptions: HashSet::new(),
surface_subscriptions: HashSet::new(),
audio_subscribed: false,
#[cfg(target_os = "linux")]
audio_bitrate_kbps: 0,
view_sizes: HashMap::new(),
scroll_offsets: HashMap::new(),
scroll_caches: HashMap::new(),
last_sent: HashMap::new(),
preview_next_send_at: HashMap::new(),
rtt_ms: 50.0,
min_rtt_ms: 50.0,
display_fps: 60.0,
delivery_bps: 262_144.0,
goodput_bps: 262_144.0,
goodput_jitter_bps: 0.0,
max_goodput_jitter_bps: 0.0,
last_goodput_sample_bps: 0.0,
avg_frame_bytes: 1_024.0,
avg_paced_frame_bytes: 1_024.0,
avg_preview_frame_bytes: 1_024.0,
avg_surface_frame_bytes: 8_192.0,
inflight_bytes: 0,
inflight_frames: VecDeque::new(),
next_send_at: Instant::now(),
probe_frames: 0.0,
frames_sent: 0,
acks_recv: 0,
acked_bytes_since_log: 0,
browser_backlog_frames: 0,
browser_ack_ahead_frames: 0,
browser_apply_ms: 0.0,
last_metrics_update: Instant::now(),
last_log: Instant::now(),
last_window_blocked_log: Instant::now(),
last_skip_log: Instant::now(),
skip_same_gen_count: 0,
skip_in_flight_count: 0,
skip_pacing_count: 0,
skip_vulkan_await_count: 0,
skip_no_subs_count: 0,
skip_not_subbed_count: 0,
skip_last_pixels_mismatch_count: 0,
encode_loop_iters: 0,
goodput_window_bytes: 0,
goodput_window_start: Instant::now(),
surface_subs: HashMap::new(),
surface_needs_keyframe: true,
surface_inflight_frames: VecDeque::new(),
vulkan_video_surfaces: HashMap::new(),
surface_view_sizes: HashMap::new(),
surface_codec_support: 0,
pressed_surface_keys: HashSet::new(),
};
(client, rx)
}
fn test_client() -> ClientState {
let (client, _rx) = test_client_with_capacity(0);
client
}
fn fill_inflight(client: &mut ClientState, frames: usize, bytes_per_frame: usize) {
let now = Instant::now();
client.inflight_bytes = frames.saturating_mul(bytes_per_frame);
client.inflight_frames = (0..frames)
.map(|_| InFlightFrame {
sent_at: now,
bytes: bytes_per_frame,
paced: true,
})
.collect();
}
fn sample_frame(text: &str) -> FrameState {
let mut frame = FrameState::new(2, 8);
frame.write_text(0, 0, text, blit_remote::CellStyle::default());
frame
}
#[test]
fn unset_view_size_accepts_zero_pair_only() {
assert!(is_unset_view_size(0, 0));
assert!(!is_unset_view_size(0, 80));
assert!(!is_unset_view_size(u16::MAX, u16::MAX));
}
#[test]
fn unsubscribe_client_from_clears_view_size() {
let mut client = test_client();
client.subscriptions.insert(7);
client.view_sizes.insert(7, (24, 80));
assert!(unsubscribe_client_from(&mut client, 7));
assert!(!client.subscriptions.contains(&7));
assert!(!client.view_sizes.contains_key(&7));
}
#[test]
fn mediated_size_uses_per_pty_view_sizes_without_lead() {
let mut session = Session::new();
let mut c1 = test_client();
let mut c2 = test_client();
c1.view_sizes.insert(7, (30, 120));
c2.view_sizes.insert(7, (24, 100));
session.clients.insert(1, c1);
session.clients.insert(2, c2);
assert_eq!(session.mediated_size_for_pty(7), Some((24, 100)));
}
#[test]
fn mediated_surface_size_picks_min_dimensions_max_scale() {
let mut session = Session::new();
let mut c1 = test_client();
let mut c2 = test_client();
c1.surface_view_sizes.insert(1, (1920, 1080, 240)); // 2×
c2.surface_view_sizes.insert(1, (1280, 720, 120)); // 1×
session.clients.insert(1, c1);
session.clients.insert(2, c2);
assert_eq!(
session.mediated_size_for_surface(1, None),
Some((1280, 720, 240))
);
}
#[test]
fn mediated_surface_size_none_when_no_clients() {
let session = Session::new();
assert_eq!(session.mediated_size_for_surface(1, None), None);
}
#[test]
fn mediated_surface_size_single_client() {
let mut session = Session::new();
let mut c1 = test_client();
c1.surface_view_sizes.insert(3, (800, 600, 120));
session.clients.insert(1, c1);
assert_eq!(
session.mediated_size_for_surface(3, None),
Some((800, 600, 120))
);
}
#[test]
fn mediated_surface_size_ignores_other_surfaces() {
let mut session = Session::new();
let mut c1 = test_client();
c1.surface_view_sizes.insert(1, (1920, 1080, 240));
c1.surface_view_sizes.insert(2, (640, 480, 120));
session.clients.insert(1, c1);
assert_eq!(
session.mediated_size_for_surface(1, None),
Some((1920, 1080, 240))
);
assert_eq!(
session.mediated_size_for_surface(2, None),
Some((640, 480, 120))
);
assert_eq!(session.mediated_size_for_surface(3, None), None);
}
#[test]
fn mediated_surface_size_clamped_to_encoder_max() {
let mut session = Session::new();
let mut c1 = test_client();
c1.surface_view_sizes.insert(1, (5000, 3000, 240));
session.clients.insert(1, c1);
assert_eq!(
session.mediated_size_for_surface(1, None),
Some((5000, 3000, 240))
);
assert_eq!(
session.mediated_size_for_surface(1, Some((3840, 2160))),
Some((3840, 2160, 240))
);
}
#[test]
fn mediated_surface_size_picks_min_across_clients() {
let mut session = Session::new();
let mut c1 = test_client();
let mut c2 = test_client();
c1.surface_view_sizes.insert(1, (1920, 1080, 120));
c2.surface_view_sizes.insert(1, (640, 360, 120));
c1.surface_subscriptions.insert(1);
c2.surface_subscriptions.insert(1);
session.clients.insert(1, c1);
session.clients.insert(2, c2);
assert_eq!(
session.mediated_size_for_surface(1, None),
Some((640, 360, 120))
);
}
#[test]
fn due_preview_reserves_the_last_lead_slot() {
let mut client = test_client();
client.lead = Some(1);
client.subscriptions.insert(1);
client.subscriptions.insert(2);
let target_frames = target_frame_window(&client);
let lead_limit = target_frames.saturating_sub(1).max(1);
fill_inflight(&mut client, lead_limit, 512);
assert!(window_open(&client));
assert!(lead_window_open(&client, false));
assert!(!lead_window_open(&client, true));
assert!(can_send_preview(&client, 2, Instant::now()));
}
#[test]
fn entering_scrollback_uses_current_visible_frame_as_baseline() {
let mut client = test_client();
let live = sample_frame("live");
client.lead = Some(7);
client.subscriptions.insert(7);
client.last_sent.insert(7, live.clone());
assert!(update_client_scroll_state(&mut client, 7, 12));
assert_eq!(client.scroll_offsets.get(&7), Some(&12));
assert_eq!(client.scroll_caches.get(&7), Some(&live));
}
#[test]
fn leaving_scrollback_seeds_live_diff_from_scrollback_view() {
let mut client = test_client();
let history = sample_frame("hist");
client.lead = Some(7);
client.subscriptions.insert(7);
client.scroll_offsets.insert(7, 12);
client.scroll_caches.insert(7, history.clone());
assert!(update_client_scroll_state(&mut client, 7, 0));
assert_eq!(client.scroll_offsets.get(&7), None);
assert_eq!(client.last_sent.get(&7), Some(&history));
assert_eq!(client.scroll_caches.get(&7), None);
}
#[tokio::test]
async fn request_surface_capture_returns_pixels_from_compositor() {
let (command_tx, command_rx) = std::sync::mpsc::channel();
std::thread::Builder::new()
.name("test-capture-reply".into())
.spawn(move || {
let CompositorCommand::Capture {
surface_id,
scale_120: _,
reply,
} = command_rx.recv().unwrap()
else {
panic!("expected capture command");
};
assert_eq!(surface_id, 7);
let _ = reply.send(Some((2, 3, vec![1, 2, 3, 4])));
})
.unwrap();
let result =
request_surface_capture_with_timeout(command_tx, 7, 0, Duration::from_millis(50)).await;
assert_eq!(result, Some((2, 3, vec![1, 2, 3, 4])));
}
#[tokio::test]
async fn request_surface_capture_returns_none_when_compositor_disconnects() {
let (command_tx, command_rx) = std::sync::mpsc::channel();
std::thread::Builder::new()
.name("test-capture-drop".into())
.spawn(move || {
let _ = command_rx.recv().unwrap();
})
.unwrap();
let result =
request_surface_capture_with_timeout(command_tx, 7, 0, Duration::from_millis(50)).await;
assert_eq!(result, None);
}
// ── frame_window ──
#[test]
fn frame_window_minimum_is_two() {
assert!(frame_window(0.0, 60.0) >= 2);
}
#[test]
fn frame_window_scales_with_rtt() {
let low = frame_window(10.0, 60.0);
let high = frame_window(200.0, 60.0);
assert!(high > low, "higher RTT should need more frames in flight");
}
#[test]
fn frame_window_scales_with_fps() {
let slow = frame_window(100.0, 10.0);
let fast = frame_window(100.0, 120.0);
assert!(fast > slow, "higher fps should need more frames in flight");
}
#[test]
fn frame_window_zero_rtt() {
assert!(frame_window(0.0, 120.0) >= 2);
}
// ── path_rtt_ms ──
#[test]
fn path_rtt_ms_uses_min_when_positive() {
let mut client = test_client();
client.rtt_ms = 100.0;
client.min_rtt_ms = 30.0;
assert_eq!(path_rtt_ms(&client), 30.0);
}
#[test]
fn path_rtt_ms_falls_back_to_rtt_when_min_zero() {
let mut client = test_client();
client.rtt_ms = 80.0;
client.min_rtt_ms = 0.0;
assert_eq!(path_rtt_ms(&client), 80.0);
}
// ── ewma_with_direction ──
#[test]
fn ewma_rising_uses_rise_alpha() {
let result = ewma_with_direction(100.0, 200.0, 0.5, 0.1);
// rise: 100 * 0.5 + 200 * 0.5 = 150
assert!((result - 150.0).abs() < 0.01);
}
#[test]
fn ewma_falling_uses_fall_alpha() {
let result = ewma_with_direction(200.0, 100.0, 0.5, 0.1);
// fall: 200 * 0.9 + 100 * 0.1 = 190
assert!((result - 190.0).abs() < 0.01);
}
#[test]
fn ewma_same_value_unchanged() {
let result = ewma_with_direction(50.0, 50.0, 0.5, 0.5);
assert!((result - 50.0).abs() < 0.01);
}
// ── advance_deadline ──
#[test]
fn advance_deadline_steps_forward() {
let now = Instant::now();
let mut deadline = now;
let interval = Duration::from_millis(16);
advance_deadline(&mut deadline, now, interval);
assert!(deadline > now);
assert!(deadline <= now + interval + Duration::from_micros(100));
}
#[test]
fn advance_deadline_resets_when_far_behind() {
let now = Instant::now();
// deadline is way in the past (more than 2 intervals ago)
let mut deadline = now - Duration::from_secs(10);
let interval = Duration::from_millis(16);
advance_deadline(&mut deadline, now, interval);
// Should snap to now + interval since scheduled + interval < now
assert!(deadline >= now);
}
#[test]
fn should_snapshot_pty_requires_dirty_and_needful() {
assert!(should_snapshot_pty(true, true, false));
assert!(!should_snapshot_pty(false, true, false));
assert!(!should_snapshot_pty(true, false, false));
}
#[test]
fn should_snapshot_pty_defers_synced_output() {
assert!(!should_snapshot_pty(true, true, true));
assert!(should_snapshot_pty(true, true, false));
}
#[test]
fn enqueue_ready_frame_refuses_new_frames_when_capped() {
let mut queue = VecDeque::new();
for cols in 1..=(READY_FRAME_QUEUE_CAP as u16) {
assert!(enqueue_ready_frame(&mut queue, FrameState::new(1, cols)));
}
assert!(!enqueue_ready_frame(
&mut queue,
FrameState::new(1, READY_FRAME_QUEUE_CAP as u16 + 1),
));
assert_eq!(queue.len(), READY_FRAME_QUEUE_CAP);
assert_eq!(queue.front().map(FrameState::cols), Some(1));
assert_eq!(
queue.back().map(FrameState::cols),
Some(READY_FRAME_QUEUE_CAP as u16),
);
}
#[test]
fn find_sync_output_end_returns_end_of_first_close_sequence() {
let bytes = b"abc\x1b[?2026lrest\x1b[?2026l";
assert_eq!(find_sync_output_end(&[], bytes), Some(11));
}
#[test]
fn find_sync_output_end_returns_none_without_close_sequence() {
assert_eq!(find_sync_output_end(&[], b"\x1b[?2026hpartial"), None);
}
#[test]
fn find_sync_output_end_detects_boundary_split_across_reads() {
assert_eq!(find_sync_output_end(b"abc\x1b[?20", b"26lrest"), Some(3));
}
#[test]
fn update_sync_scan_tail_keeps_recent_suffix_only() {
let mut tail = Vec::new();
update_sync_scan_tail(&mut tail, b"123456789");
assert_eq!(tail, b"3456789");
}
// ── window_saturated ──
#[test]
fn window_saturated_at_90_percent_frames() {
let client = test_client();
let target = target_frame_window(&client);
let frames_90 = (target * 9).div_ceil(10); // ceil(target * 0.9)
assert!(window_saturated(&client, frames_90, 0));
}
#[test]
fn window_saturated_not_at_low_usage() {
let client = test_client();
assert!(!window_saturated(&client, 1, 0));
}
#[test]
fn window_saturated_at_90_percent_bytes() {
let client = test_client();
let target_bytes = target_byte_window(&client);
let bytes_90 = (target_bytes * 9).div_ceil(10);
assert!(window_saturated(&client, 0, bytes_90));
}
// ── outbox_queued_frames / outbox_backpressured ──
#[test]
fn outbox_queued_frames_zero_when_empty() {
let client = test_client();
assert_eq!(outbox_queued_frames(&client), 0);
}
#[test]
fn outbox_backpressured_when_queue_full() {
let (client, _rx) = test_client_with_capacity(0);
// Fill the channel to trigger backpressure
for _ in 0..OUTBOX_SOFT_QUEUE_LIMIT_FRAMES {
let _ = send_outbox(&client, vec![0u8]);
}
assert!(outbox_backpressured(&client));
}
#[test]
fn outbox_not_backpressured_by_small_frames_under_byte_budget() {
let (client, _rx) = test_client_with_capacity(0);
for _ in 0..(OUTBOX_SOFT_QUEUE_LIMIT_FRAMES - 1) {
let _ = send_outbox(&client, vec![0u8; 512]);
}
assert!(!outbox_backpressured(&client));
}
#[test]
fn outbox_backpressured_by_large_queued_bytes() {
let (client, _rx) = test_client_with_capacity(0);
// First frame is always allowed through, even at the byte limit, so
// pending encoders can make progress when keyframes exceed the cap.
let _ = send_outbox(&client, vec![0u8; OUTBOX_SOFT_QUEUE_LIMIT_BYTES]);
assert!(!outbox_backpressured(&client));
// A second frame pushes total bytes past the soft limit.
let _ = send_outbox(&client, vec![0u8; 1]);
assert!(outbox_backpressured(&client));
}
#[test]
fn outbox_not_backpressured_when_empty() {
let client = test_client();
assert!(!outbox_backpressured(&client));
}
// ── browser_pacing_fps baseline ──
#[test]
fn browser_pacing_fps_matches_display_fps_when_browser_ready() {
let mut client = test_client();
client.rtt_ms = 1.0;
client.min_rtt_ms = 1.0;
client.browser_backlog_frames = 0;
client.browser_ack_ahead_frames = 0;
client.browser_apply_ms = 0.0;
client.goodput_bps = 1_000_000.0;
client.delivery_bps = 1_000_000.0;
client.display_fps = 144.0;
assert!((browser_pacing_fps(&client) - 144.0).abs() < 0.01);
}
#[test]
fn browser_pacing_fps_drops_below_display_fps_when_backlogged() {
let mut client = test_client();
client.browser_backlog_frames = 20;
let fps = browser_pacing_fps(&client);
assert!(fps >= 1.0);
assert!(fps < client.display_fps);
}
// ── effective_rtt_ms ──
#[test]
fn effective_rtt_ms_equals_path_when_queue_is_empty() {
let mut client = test_client();
client.rtt_ms = 1.0;
client.min_rtt_ms = 1.0;
client.browser_backlog_frames = 0;
client.browser_ack_ahead_frames = 0;
client.browser_apply_ms = 0.0;
client.goodput_bps = 1_000_000.0;
client.delivery_bps = 1_000_000.0;
assert!((effective_rtt_ms(&client) - 1.0).abs() < 0.01);
}
#[test]
fn effective_rtt_ms_at_least_path_rtt() {
let client = test_client();
assert!(effective_rtt_ms(&client) >= path_rtt_ms(&client));
}
// ── target_frame_window ──
#[test]
fn target_frame_window_at_least_two() {
let client = test_client();
assert!(target_frame_window(&client) >= 2);
}
#[test]
fn target_frame_window_grows_with_probe() {
let mut client = test_client();
let base = target_frame_window(&client);
client.probe_frames = 10.0;
let probed = target_frame_window(&client);
assert!(probed > base, "probe_frames should grow the window");
}
// ── bandwidth_floor_bps ──
#[test]
fn bandwidth_floor_bps_at_least_16k() {
let mut client = test_client();
client.goodput_bps = 0.0;
client.delivery_bps = 0.0;
assert_eq!(bandwidth_floor_bps(&client), 0.0);
}
#[test]
fn bandwidth_floor_bps_scales_with_goodput() {
let mut client = test_client();
client.goodput_bps = 1_000_000.0;
client.delivery_bps = 1_000_000.0;
let floor = bandwidth_floor_bps(&client);
assert!(floor > 0.0);
}
#[test]
fn browser_ready_delivery_floor_can_drive_large_frames_to_display_fps() {
let mut client = test_client();
client.display_fps = 60.0;
client.browser_backlog_frames = 0;
client.browser_ack_ahead_frames = 0;
client.browser_apply_ms = 0.2;
client.goodput_bps = 3_000_000.0;
client.delivery_bps = 9_500_000.0;
client.last_goodput_sample_bps = 3_000_000.0;
client.avg_paced_frame_bytes = 150_000.0;
client.avg_preview_frame_bytes = 1_024.0;
client.avg_frame_bytes = 150_000.0;
assert!(
(pacing_fps(&client) - client.display_fps).abs() < 0.01,
"browser-ready delivery floor should let large frames reach display_fps on a fast path",
);
}
// ── pacing_fps ──
#[test]
fn pacing_fps_zero_when_no_bandwidth() {
let mut client = test_client();
client.goodput_bps = 0.0;
client.delivery_bps = 0.0;
client.last_goodput_sample_bps = 0.0;
assert!(
pacing_fps(&client) == 0.0,
"pacing_fps should be 0 with zero bandwidth"
);
}
#[test]
fn pacing_fps_reaches_display_fps_when_not_bandwidth_limited() {
let mut client = test_client();
client.rtt_ms = 1.0;
client.min_rtt_ms = 1.0;
client.browser_backlog_frames = 0;
client.browser_ack_ahead_frames = 0;
client.browser_apply_ms = 0.0;
client.goodput_bps = 1_000_000.0;
client.delivery_bps = 1_000_000.0;
client.display_fps = 60.0;
assert!((pacing_fps(&client) - 60.0).abs() < 0.01);
}
// ── throughput_limited ──
#[test]
fn throughput_limited_when_low_bandwidth() {
let mut client = test_client();
client.goodput_bps = 1_000.0;
client.delivery_bps = 1_000.0;
client.last_goodput_sample_bps = 0.0;
assert!(throughput_limited(&client));
}
#[test]
fn throughput_not_limited_with_high_bandwidth() {
let mut client = test_client();
client.goodput_bps = 100_000_000.0;
client.delivery_bps = 100_000_000.0;
assert!(!throughput_limited(&client));
}
// ── browser_pacing_fps ──
#[test]
fn browser_pacing_fps_at_least_one() {
let client = test_client();
assert!(browser_pacing_fps(&client) >= 1.0);
}
#[test]
fn browser_pacing_fps_reduced_by_high_backlog() {
let mut client = test_client();
let normal = browser_pacing_fps(&client);
client.browser_backlog_frames = 20;
let backlogged = browser_pacing_fps(&client);
assert!(backlogged < normal, "high backlog should reduce pacing fps");
}
#[test]
fn browser_pacing_fps_reduced_by_high_ack_ahead() {
let mut client = test_client();
let normal = browser_pacing_fps(&client);
client.browser_ack_ahead_frames = 10;
let ahead = browser_pacing_fps(&client);
assert!(ahead < normal, "high ack_ahead should reduce pacing fps");
}
// ── browser_backlog_blocked ──
#[test]
fn browser_backlog_blocked_over_threshold() {
let mut client = test_client();
client.browser_backlog_frames = 9;
assert!(browser_backlog_blocked(&client));
}
#[test]
fn browser_backlog_not_blocked_under_threshold() {
let mut client = test_client();
client.browser_backlog_frames = 8;
assert!(!browser_backlog_blocked(&client));
}
// ── byte_budget_for ──
#[test]
fn byte_budget_for_at_least_one_frame() {
let client = test_client();
let budget = byte_budget_for(&client, 10.0);
assert!(budget >= client.avg_frame_bytes.max(256.0) as usize);
}
#[test]
fn byte_budget_for_grows_with_time() {
let client = test_client();
let short = byte_budget_for(&client, 10.0);
let long = byte_budget_for(&client, 1000.0);
assert!(long >= short);
}
// ── target_byte_window ──
#[test]
fn target_byte_window_positive() {
let client = test_client();
assert!(target_byte_window(&client) > 0);
}
#[test]
fn target_byte_window_covers_frame_window() {
let client = test_client();
let byte_win = target_byte_window(&client);
let frame_win = target_frame_window(&client);
let min_bytes =
(client.avg_paced_frame_bytes.max(256.0) * frame_win.max(2) as f32).ceil() as usize;
assert!(
byte_win >= min_bytes,
"byte window should cover at least frame_window worth of paced frames"
);
}
// ── send_interval ──
#[test]
fn send_interval_matches_browser_pacing() {
let client = test_client();
let interval = send_interval(&client);
let expected = Duration::from_secs_f64(1.0 / browser_pacing_fps(&client) as f64);
let diff = interval.abs_diff(expected);
assert!(diff < Duration::from_micros(10));
}
// ── preview_fps ──
#[test]
fn preview_fps_at_least_one() {
let client = test_client();
assert!(preview_fps(&client) >= 1.0);
}
// ── window_open ──
#[test]
fn window_open_initially() {
let client = test_client();
assert!(window_open(&client));
}
#[test]
fn window_open_false_when_browser_blocked() {
let mut client = test_client();
client.browser_backlog_frames = 20;
assert!(!window_open(&client));
}
#[test]
fn window_open_false_when_inflight_full() {
let mut client = test_client();
let target = target_frame_window(&client);
fill_inflight(&mut client, target + 10, 1024);
assert!(!window_open(&client));
}
// ── lead_window_open ──
#[test]
fn lead_window_open_no_reserve_same_as_window_open() {
let client = test_client();
assert_eq!(lead_window_open(&client, false), window_open(&client));
}
#[test]
fn lead_window_open_reserves_preview_slot() {
let mut client = test_client();
client.lead = Some(1);
client.subscriptions.insert(1);
let target = target_frame_window(&client);
// Fill to just under target minus reserve
fill_inflight(&mut client, target.saturating_sub(1), 512);
// Without reserve: may still be open
// With reserve: should be closed
assert!(!lead_window_open(&client, true));
}
// ── can_send_frame ──
#[test]
fn can_send_frame_when_window_open_and_time_due() {
let mut client = test_client();
client.next_send_at = Instant::now() - Duration::from_millis(100);
assert!(can_send_frame(&client, Instant::now(), false));
}
#[test]
fn can_send_frame_false_when_not_due() {
let mut client = test_client();
client.next_send_at = Instant::now() + Duration::from_secs(10);
assert!(!can_send_frame(&client, Instant::now(), false));
}
#[test]
fn can_send_frame_false_when_window_closed() {
let mut client = test_client();
client.browser_backlog_frames = 20; // triggers browser_backlog_blocked
client.next_send_at = Instant::now() - Duration::from_millis(100);
assert!(!can_send_frame(&client, Instant::now(), false));
}
// ── record_send / record_ack state transitions ──
#[test]
fn record_send_increases_inflight() {
let mut client = test_client();
let now = Instant::now();
assert_eq!(client.inflight_bytes, 0);
assert_eq!(client.inflight_frames.len(), 0);
record_send(&mut client, 1000, now, true);
assert_eq!(client.inflight_bytes, 1000);
assert_eq!(client.inflight_frames.len(), 1);
record_send(&mut client, 500, now, false);
assert_eq!(client.inflight_bytes, 1500);
assert_eq!(client.inflight_frames.len(), 2);
}
#[test]
fn record_send_paced_advances_deadline() {
let mut client = test_client();
let now = Instant::now();
client.next_send_at = now;
record_send(&mut client, 1000, now, true);
assert!(client.next_send_at > now);
}
#[test]
fn record_send_unpaced_does_not_advance_deadline() {
let mut client = test_client();
let now = Instant::now();
let before = client.next_send_at;
record_send(&mut client, 1000, now, false);
assert_eq!(client.next_send_at, before);
}
#[test]
fn record_ack_decreases_inflight() {
let mut client = test_client();
let now = Instant::now();
record_send(&mut client, 1000, now, true);
record_send(&mut client, 500, now, true);
assert_eq!(client.inflight_frames.len(), 2);
record_ack(&mut client);
assert_eq!(client.inflight_frames.len(), 1);
assert_eq!(client.inflight_bytes, 500);
}
#[test]
fn record_ack_on_empty_clears_bytes() {
let mut client = test_client();
client.inflight_bytes = 999; // stale state
record_ack(&mut client);
assert_eq!(client.inflight_bytes, 0);
}
#[test]
fn record_ack_updates_rtt_estimate() {
let mut client = test_client();
let now = Instant::now();
client.inflight_frames.push_back(InFlightFrame {
sent_at: now - Duration::from_millis(20),
bytes: 512,
paced: true,
});
client.inflight_bytes = 512;
let old_rtt = client.rtt_ms;
record_ack(&mut client);
// RTT should have been updated (moved toward ~20ms from the default 50ms)
assert!(
(client.rtt_ms - old_rtt).abs() > 0.01,
"rtt_ms should be updated after ack"
);
}
#[test]
fn record_ack_paced_updates_avg_paced_frame_bytes() {
let mut client = test_client();
let now = Instant::now();
client.inflight_frames.push_back(InFlightFrame {
sent_at: now - Duration::from_millis(10),
bytes: 4096,
paced: true,
});
client.inflight_bytes = 4096;
let old_avg = client.avg_paced_frame_bytes;
record_ack(&mut client);
// Should move toward 4096 from 1024
assert!(client.avg_paced_frame_bytes > old_avg);
}
#[test]
fn record_ack_unpaced_updates_avg_preview_frame_bytes() {
let mut client = test_client();
let now = Instant::now();
client.inflight_frames.push_back(InFlightFrame {
sent_at: now - Duration::from_millis(10),
bytes: 8192,
paced: false,
});
client.inflight_bytes = 8192;
let old_avg = client.avg_preview_frame_bytes;
record_ack(&mut client);
assert!(client.avg_preview_frame_bytes > old_avg);
}
// ── Session::pty_list_msg format ──
#[test]
fn pty_list_msg_empty_session() {
let sess = Session::new();
let msg = sess.pty_list_msg();
assert_eq!(msg[0], S2C_LIST);
assert_eq!(u16::from_le_bytes([msg[1], msg[2]]), 0);
assert_eq!(msg.len(), 3);
}
#[test]
fn pty_list_msg_includes_tags() {
let _sess = Session::new();
// Insert minimal Pty entries. We can't call spawn_pty, so build
// a mock-like Pty with a stub driver. Instead, directly insert
// into the HashMap using an unsafe-free approach: just build the
// wire message by hand and verify against a known layout.
//
// The wire format is: [S2C_LIST] [count:u16le] [id:u16le tag_len:u16le tag_bytes]...
//
// Since we can't easily construct a Pty without forking, verify
// the format by constructing the expected bytes and comparing.
let tag1 = "shell";
let tag2 = "build";
// Expected wire for ptys {1 => "shell", 3 => "build"} sorted by id:
let mut expected = vec![S2C_LIST];
expected.extend_from_slice(&2u16.to_le_bytes());
// id=1
expected.extend_from_slice(&1u16.to_le_bytes());
expected.extend_from_slice(&(tag1.len() as u16).to_le_bytes());
expected.extend_from_slice(tag1.as_bytes());
// id=3
expected.extend_from_slice(&3u16.to_le_bytes());
expected.extend_from_slice(&(tag2.len() as u16).to_le_bytes());
expected.extend_from_slice(tag2.as_bytes());
// Verify our expected format starts with S2C_LIST and has correct count
assert_eq!(expected[0], S2C_LIST);
assert_eq!(u16::from_le_bytes([expected[1], expected[2]]), 2);
// Verify tags are embedded
let msg_str = String::from_utf8_lossy(&expected);
assert!(msg_str.contains("shell"));
assert!(msg_str.contains("build"));
}
// ── can_send_preview / record_preview_send ──
#[test]
fn can_send_preview_true_when_due() {
let mut client = test_client();
let now = Instant::now();
client
.preview_next_send_at
.insert(5, now - Duration::from_millis(100));
assert!(can_send_preview(&client, 5, now));
}
#[test]
fn can_send_preview_false_when_not_due() {
let mut client = test_client();
let now = Instant::now();
client
.preview_next_send_at
.insert(5, now + Duration::from_secs(10));
assert!(!can_send_preview(&client, 5, now));
}
#[test]
fn can_send_preview_false_when_window_closed() {
let mut client = test_client();
client.browser_backlog_frames = 20;
let now = Instant::now();
assert!(!can_send_preview(&client, 5, now));
}
#[test]
fn can_send_preview_true_for_unseen_pid() {
let client = test_client();
let now = Instant::now();
// No entry in preview_next_send_at means deadline defaults to now
assert!(can_send_preview(&client, 99, now));
}
#[test]
fn record_preview_send_sets_future_deadline() {
let mut client = test_client();
let now = Instant::now();
record_preview_send(&mut client, 5, now);
let deadline = client.preview_next_send_at.get(&5).unwrap();
assert!(*deadline > now);
}
#[test]
fn record_preview_send_successive_calls_advance() {
let mut client = test_client();
let now = Instant::now();
record_preview_send(&mut client, 5, now);
let first = *client.preview_next_send_at.get(&5).unwrap();
record_preview_send(&mut client, 5, first);
let second = *client.preview_next_send_at.get(&5).unwrap();
assert!(second > first, "successive sends should advance deadline");
}
// ── congestion control end-to-end properties ──
//
// These tests encode the two goals of the congestion controller:
// 1. Browser-ready, well-provisioned path → full display FPS, minimal added latency
// 2. Bottleneck → lowest sustainable FPS, fast recovery when pipe clears
//
// Some tests assert desired future behaviour and currently FAIL due to
// known issues (min_rtt contamination, lead_floor dominating byte window).
// They are marked with a comment so they are easy to find when fixing.
/// Return a client in ideal low-latency, high-bandwidth conditions:
/// browser ready, abundant bandwidth, and tiny RTT. The normal pacing path
/// should still reach display_fps.
fn browser_ready_high_bandwidth_client() -> ClientState {
let mut c = test_client();
c.display_fps = 120.0;
c.rtt_ms = 1.0;
c.min_rtt_ms = 1.0;
c.goodput_bps = 50_000_000.0;
c.delivery_bps = 50_000_000.0;
c.last_goodput_sample_bps = 50_000_000.0;
c.avg_paced_frame_bytes = 30_000.0;
c.avg_preview_frame_bytes = 1_024.0;
c.avg_frame_bytes = 30_000.0;
c.browser_apply_ms = 0.3;
c
}
/// Return a client that has converged to a clearly congested state:
/// ~10× min_rtt inflation, low goodput.
fn congested_client() -> ClientState {
let mut c = test_client();
c.display_fps = 120.0;
c.rtt_ms = 500.0;
c.min_rtt_ms = 40.0;
c.goodput_bps = 200_000.0;
c.delivery_bps = 150_000.0;
c.last_goodput_sample_bps = 200_000.0;
c.avg_paced_frame_bytes = 50_000.0;
c.avg_preview_frame_bytes = 1_024.0;
c.avg_frame_bytes = 50_000.0;
c.goodput_jitter_bps = 50_000.0;
c.max_goodput_jitter_bps = 200_000.0;
c.browser_apply_ms = 1.0;
c
}
/// Simulate one ACK: insert a frame with the given RTT into inflight and
/// call record_ack. Forces a goodput-window sample each call so that
/// goodput estimates respond within a few calls.
fn sim_ack(client: &mut ClientState, bytes: usize, rtt_ms: f32) {
let sent_at = Instant::now() - Duration::from_millis(rtt_ms as u64);
client.inflight_bytes += bytes;
client.inflight_frames.push_back(InFlightFrame {
sent_at,
bytes,
paced: true,
});
// Age the goodput window so record_ack always emits a sample.
client.goodput_window_start = Instant::now() - Duration::from_millis(25);
record_ack(client);
}
fn sim_acks(client: &mut ClientState, n: usize, bytes: usize, rtt_ms: f32) {
for _ in 0..n {
sim_ack(client, bytes, rtt_ms);
}
}
// ── property: full FPS on a browser-ready path ──
#[test]
fn browser_ready_high_bandwidth_client_uses_full_display_fps() {
let client = browser_ready_high_bandwidth_client();
assert!(
(pacing_fps(&client) - client.display_fps).abs() < 0.01,
"pacing_fps {} should equal display_fps {} when browser is ready and bandwidth is abundant",
pacing_fps(&client),
client.display_fps,
);
}
#[test]
fn browser_ready_high_bandwidth_client_send_interval_within_one_frame() {
let client = browser_ready_high_bandwidth_client();
let interval_ms = send_interval(&client).as_secs_f32() * 1000.0;
let frame_ms = 1000.0 / client.display_fps;
assert!(
interval_ms <= frame_ms + 0.1,
"send_interval {interval_ms:.2}ms exceeds one frame ({frame_ms:.2}ms) when browser is ready"
);
}
// ── property: degraded FPS when bottlenecked ──
#[test]
fn congested_pipe_reduces_pacing_fps_substantially() {
let client = congested_client();
let fps = pacing_fps(&client);
assert!(
fps < client.display_fps * 0.5,
"pacing_fps {fps:.0} should be well below display_fps {} when congested",
client.display_fps,
);
}
#[test]
fn congested_pipe_is_throughput_limited() {
let client = congested_client();
assert!(
throughput_limited(&client),
"congested client must be recognised as throughput-limited"
);
}
// ── property: byte window should stay near BDP ──
//
// KNOWN FAILING: lead_floor in target_byte_window overrides the BDP
// budget when avg_paced_frame_bytes is large. Fix: cap lead_floor.
#[test]
fn byte_window_bounded_near_bdp_when_congested() {
let client = congested_client();
// BDP at the unloaded path RTT.
let bdp = client.goodput_bps * (path_rtt_ms(&client) / 1_000.0);
let window = target_byte_window(&client);
assert!(
window < bdp as usize * 8,
"byte window {window}B is {:.1}× BDP ({bdp:.0}B); \
expected ≤ 8× — lead_floor may be dominating",
window as f32 / bdp.max(1.0),
);
}
// ── property: min_rtt must not drift upward under congestion ──
//
// KNOWN FAILING: the `min_rtt_ms * 0.999 + rtt_ms * 0.001` update
// bleeds queued RTT into min_rtt.
#[test]
fn min_rtt_not_contaminated_by_congested_rtts() {
let mut client = test_client();
client.display_fps = 120.0;
client.rtt_ms = 40.0;
client.min_rtt_ms = 40.0;
client.goodput_bps = 2_000_000.0;
client.delivery_bps = 2_000_000.0;
client.avg_paced_frame_bytes = 30_000.0;
client.avg_preview_frame_bytes = 1_024.0;
let original_min = client.min_rtt_ms;
// 200 ACKs arriving with 500ms RTT (severe congestion).
sim_acks(&mut client, 200, 30_000, 500.0);
assert!(
client.min_rtt_ms < original_min * 2.0,
"min_rtt drifted from {original_min}ms to {:.1}ms after 200 congested ACKs",
client.min_rtt_ms,
);
}
// ── property: fast recovery when congestion clears ──
#[test]
fn delivery_bps_rises_quickly_when_congestion_clears() {
let mut client = congested_client();
let before = client.delivery_bps;
// 10 ACKs at low latency / high throughput.
sim_acks(&mut client, 10, 30_000, 40.0);
assert!(
client.delivery_bps > before * 2.0,
"delivery_bps {:.0} should more than double from {before:.0} after 10 fast ACKs",
client.delivery_bps,
);
}
#[test]
fn pacing_fps_recovers_after_congestion_clears() {
let mut client = congested_client();
// Use window-saturated rounds: fill the window with frames, age the
// goodput window once, then ACK all. The first ACK each round emits
// a sample; the remaining target-1 ACKs carry over into the next
// window, so sample throughput grows as target grows — mimicking a
// real link where the sender keeps the pipe full across one RTT.
for _ in 0..40 {
let target = target_frame_window(&client).max(2);
for _ in 0..target {
let sent_at = Instant::now() - Duration::from_millis(40);
client.inflight_bytes += 30_000;
client.inflight_frames.push_back(InFlightFrame {
sent_at,
bytes: 30_000,
paced: true,
});
}
client.goodput_window_start = Instant::now() - Duration::from_millis(25);
for _ in 0..target {
record_ack(&mut client);
}
}
let fps = pacing_fps(&client);
assert!(
fps > client.display_fps * 0.7,
"pacing_fps {fps:.0} didn't recover toward display_fps {} \
after window-saturated rounds at low RTT",
client.display_fps,
);
}
#[test]
fn rtt_estimate_drops_quickly_when_congestion_clears() {
let mut client = test_client();
client.rtt_ms = 500.0;
client.min_rtt_ms = 40.0;
client.goodput_bps = 2_000_000.0;
client.avg_paced_frame_bytes = 30_000.0;
client.avg_preview_frame_bytes = 1_024.0;
// The asymmetric EWMA uses rise=0.125, fall=0.25, so rtt_ms drops
// at fall_alpha=0.25 per sample toward the new low.
sim_acks(&mut client, 10, 30_000, 40.0);
assert!(
client.rtt_ms < 300.0,
"rtt_ms {:.0}ms did not fall fast enough after congestion cleared",
client.rtt_ms,
);
}
// ── property: probing ──
#[test]
fn probe_collapses_immediately_on_queue_delay() {
let mut client = test_client();
client.display_fps = 120.0;
client.rtt_ms = 40.0;
client.min_rtt_ms = 40.0;
client.goodput_bps = 5_000_000.0;
client.delivery_bps = 5_000_000.0;
client.last_goodput_sample_bps = 5_000_000.0;
client.avg_paced_frame_bytes = 10_000.0;
client.avg_preview_frame_bytes = 1_024.0;
client.probe_frames = 10.0;
// ACKs arriving with high RTT signal queue buildup.
sim_acks(&mut client, 5, 10_000, 600.0);
assert!(
client.probe_frames < 5.0,
"probe_frames {:.1} should have collapsed on queue delay signal",
client.probe_frames,
);
}
#[test]
fn probe_grows_when_window_saturated_with_clean_rtt() {
let mut client = test_client();
client.display_fps = 120.0;
client.rtt_ms = 40.0;
client.min_rtt_ms = 40.0;
client.goodput_bps = 5_000_000.0;
client.delivery_bps = 5_000_000.0;
client.last_goodput_sample_bps = 5_000_000.0;
client.avg_paced_frame_bytes = 10_000.0;
client.avg_preview_frame_bytes = 1_024.0;
client.goodput_jitter_bps = 0.0;
client.max_goodput_jitter_bps = 0.0;
client.probe_frames = 0.0;
// Saturate inflight so window_saturated returns true during acks.
let target = target_frame_window(&client);
for _ in 0..target {
let sent_at = Instant::now() - Duration::from_millis(40);
client.inflight_bytes += 10_000;
client.inflight_frames.push_back(InFlightFrame {
sent_at,
bytes: 10_000,
paced: true,
});
}
// Ack one frame with clean RTT. One saturated ACK is sufficient to
// verify the property: as probe_frames increments, target_frame_window
// grows, so the remaining (target-1) frames would fall below the 90%
// threshold and trigger gentle decay. The property under test is that
// *receiving an ACK while window-saturated* increments probe_frames —
// not that it stays incremented across subsequent unsaturated ACKs.
// Also: do NOT age the goodput window — that would emit a per-frame
// sample far below goodput_bps, spiking jitter and collapsing probe.
record_ack(&mut client);
assert!(
client.probe_frames > 0.0,
"probe_frames should grow when window-saturated with clean RTT"
);
}
// ── property: frame window larger on high-latency links ──
#[test]
fn frame_window_larger_on_high_latency_link() {
let mut lo = test_client();
lo.display_fps = 120.0;
lo.rtt_ms = 10.0;
lo.min_rtt_ms = 10.0;
lo.goodput_bps = 5_000_000.0;
lo.delivery_bps = 5_000_000.0;
lo.avg_paced_frame_bytes = 10_000.0;
lo.avg_preview_frame_bytes = 1_024.0;
let mut hi = test_client();
hi.display_fps = 120.0;
hi.rtt_ms = 200.0;
hi.min_rtt_ms = 200.0;
hi.goodput_bps = 5_000_000.0;
hi.delivery_bps = 5_000_000.0;
hi.avg_paced_frame_bytes = 10_000.0;
hi.avg_preview_frame_bytes = 1_024.0;
let lo_win = target_frame_window(&lo);
let hi_win = target_frame_window(&hi);
assert!(
hi_win > lo_win,
"high-latency link ({hi_win}f) should need more frames in flight \
than low-latency ({lo_win}f)"
);
}
// ── property: small-frame byte window allows pipelining ──
#[test]
fn small_frame_byte_window_enables_pipelining() {
// Tiny terminal frames (~1KB) with a stale congested RTT and low
// goodput estimate (stop-and-wait artifact): byte window must be at
// least target_frame_window × frame_bytes so the sender can pipeline
// rather than stay stuck in stop-and-wait.
let mut client = test_client();
client.display_fps = 120.0;
client.rtt_ms = 165.0;
client.min_rtt_ms = 8.0;
client.goodput_bps = 11_000.0; // stop-and-wait artifact
client.delivery_bps = 6_800.0;
client.last_goodput_sample_bps = 11_000.0;
client.avg_paced_frame_bytes = 1_120.0;
client.avg_preview_frame_bytes = 1_024.0;
client.goodput_jitter_bps = 4_300.0;
client.max_goodput_jitter_bps = 6_500.0;
let window = target_byte_window(&client);
let frames = target_frame_window(&client);
let pipeline = frames * 1_120;
assert!(
window >= pipeline,
"byte window {window}B should be >= pipeline ({frames}f × 1120B = {pipeline}B) \
so small frames can pipeline across the RTT"
);
}
#[test]
fn large_frame_byte_window_bounded_by_one_frame_floor() {
// With large frames (50KB), pipelining the full frame window (5×50KB=250KB)
// would be many multiples of BDP. Byte window should fall back to
// the one-frame floor so the BDP budget governs.
let mut client = test_client();
client.display_fps = 120.0;
client.rtt_ms = 165.0;
client.min_rtt_ms = 8.0;
client.goodput_bps = 11_000.0;
client.delivery_bps = 6_800.0;
client.last_goodput_sample_bps = 11_000.0;
client.avg_paced_frame_bytes = 50_000.0; // large frame
client.avg_preview_frame_bytes = 1_024.0;
client.goodput_jitter_bps = 0.0;
client.max_goodput_jitter_bps = 0.0;
let window = target_byte_window(&client);
let frames = target_frame_window(&client);
let pipeline = frames.saturating_mul(50_000);
assert!(
window < pipeline,
"byte window {window}B should be < full pipeline {pipeline}B \
({frames}f × 50KB) — large frames must use one-frame floor"
);
assert!(
window >= 50_000,
"byte window {window}B must be at least one frame (50KB)"
);
}
// ── property: preview reservation applies uniformly ──
#[test]
fn preview_reservation_applies_even_on_low_latency_high_bandwidth_links() {
let mut client = browser_ready_high_bandwidth_client();
client.lead = Some(1);
client.subscriptions.insert(1);
let target = target_frame_window(&client);
fill_inflight(&mut client, target.saturating_sub(1), 512);
assert!(
!lead_window_open(&client, true),
"preview reservation should apply uniformly for lead clients"
);
}
// ── property: blip recovery on healthy paths ──
#[test]
fn probe_recovers_on_healthy_path_after_blip() {
let mut client = browser_ready_high_bandwidth_client();
client.probe_frames = 8.0;
// Blip: 3 ACKs with inflated RTT crush probes.
sim_acks(&mut client, 3, 30_000, 200.0);
let post_blip = client.probe_frames;
assert!(
post_blip < 4.0,
"probe_frames {post_blip:.1} should have dropped after blip"
);
// Reset browser metrics to healthy (browser cleared backlog).
client.browser_backlog_frames = 0;
client.browser_ack_ahead_frames = 0;
client.browser_apply_ms = 0.3;
// Recovery: 20 healthy ACKs at low RTT on an underfilled path.
sim_acks(&mut client, 20, 30_000, 1.0);
assert!(
client.probe_frames > post_blip,
"probe_frames {:.1} should have recovered from {post_blip:.1} after healthy ACKs",
client.probe_frames,
);
}
#[test]
fn jitter_decays_fast_on_browser_ready_path() {
let mut client = browser_ready_high_bandwidth_client();
// Inject elevated jitter (simulating post-blip state).
client.max_goodput_jitter_bps = client.goodput_bps * 0.4;
client.goodput_jitter_bps = client.goodput_bps * 0.3;
let initial_jitter = client.max_goodput_jitter_bps;
// 10 healthy ACKs on a browser-ready path.
sim_acks(&mut client, 10, 30_000, 1.0);
assert!(
client.max_goodput_jitter_bps < initial_jitter * 0.5,
"max_goodput_jitter_bps {:.0} should have decayed below {:.0} \
(50% of initial {initial_jitter:.0}) after 10 healthy ACKs on a ready path",
client.max_goodput_jitter_bps,
initial_jitter * 0.5,
);
}
#[test]
fn byte_budget_uses_floor_when_goodput_depressed() {
let mut client = browser_ready_high_bandwidth_client();
client.goodput_bps = 100_000.0;
let budget = byte_budget_for(&client, 100.0);
let floor_budget = (bandwidth_floor_bps(&client) * 100.0 / 1_000.0).ceil() as usize;
assert!(
budget >= floor_budget,
"byte_budget {budget} should be at least bandwidth_floor-based {floor_budget} \
when goodput_bps is depressed but delivery_bps is high"
);
}
#[test]
fn probe_floor_maintained_under_congestion_signal() {
let mut client = test_client();
client.display_fps = 120.0;
client.rtt_ms = 40.0;
client.min_rtt_ms = 40.0;
client.goodput_bps = 5_000_000.0;
client.delivery_bps = 5_000_000.0;
client.last_goodput_sample_bps = 5_000_000.0;
client.avg_paced_frame_bytes = 10_000.0;
client.avg_preview_frame_bytes = 1_024.0;
client.probe_frames = 10.0;
// Many ACKs with high RTT: probes should not drop below the floor.
sim_acks(&mut client, 20, 10_000, 600.0);
assert!(
client.probe_frames >= 1.0,
"probe_frames {:.1} should not drop below the floor of 1.0",
client.probe_frames,
);
}
// ── parse_terminal_queries ──
#[test]
fn parse_tq_da1_bare() {
let results = parse_terminal_queries(b"\x1b[c", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert!(results[0].starts_with("\x1b[?64;"));
}
#[test]
fn parse_tq_da1_with_zero_param() {
let results = parse_terminal_queries(b"\x1b[0c", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert!(results[0].starts_with("\x1b[?64;"));
}
#[test]
fn parse_tq_dsr_cursor_position() {
let results = parse_terminal_queries(b"\x1b[6n", (24, 80), (5, 10));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b[6;11R");
}
#[test]
fn parse_tq_dsr_status() {
let results = parse_terminal_queries(b"\x1b[5n", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b[0n");
}
#[test]
fn parse_tq_window_size_cells() {
let results = parse_terminal_queries(b"\x1b[18t", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b[8;24;80t");
}
#[test]
fn parse_tq_window_size_pixels() {
let results = parse_terminal_queries(b"\x1b[14t", (30, 100), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b[4;480;800t");
}
#[test]
fn parse_tq_multiple_queries() {
let data = b"\x1b[c\x1b[6n\x1b[5n";
let results = parse_terminal_queries(data, (24, 80), (2, 3));
assert_eq!(results.len(), 3);
assert!(results[0].starts_with("\x1b[?64;"));
assert_eq!(results[1], "\x1b[3;4R");
assert_eq!(results[2], "\x1b[0n");
}
#[test]
fn parse_tq_question_mark_sequences_skipped() {
let results = parse_terminal_queries(b"\x1b[?1h", (24, 80), (0, 0));
assert!(results.is_empty());
}
#[test]
fn parse_tq_unknown_final_byte_ignored() {
let results = parse_terminal_queries(b"\x1b[42z", (24, 80), (0, 0));
assert!(results.is_empty());
}
#[test]
fn parse_tq_empty_input() {
let results = parse_terminal_queries(b"", (24, 80), (0, 0));
assert!(results.is_empty());
}
#[test]
fn parse_tq_plain_text_no_csi() {
let results = parse_terminal_queries(b"hello world", (24, 80), (0, 0));
assert!(results.is_empty());
}
#[test]
fn parse_tq_interleaved_with_text() {
let results = parse_terminal_queries(b"abc\x1b[cdef\x1b[6n", (24, 80), (1, 2));
assert_eq!(results.len(), 2);
}
// ── parse_terminal_queries: OSC ──
#[test]
fn parse_tq_osc11_background_color_bel() {
let results = parse_terminal_queries(b"\x1b]11;?\x07", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b]11;rgb:0000/0000/0000\x1b\\");
}
#[test]
fn parse_tq_osc11_background_color_st() {
let results = parse_terminal_queries(b"\x1b]11;?\x1b\\", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b]11;rgb:0000/0000/0000\x1b\\");
}
#[test]
fn parse_tq_osc10_foreground_color() {
let results = parse_terminal_queries(b"\x1b]10;?\x07", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b]10;rgb:ffff/ffff/ffff\x1b\\");
}
#[test]
fn parse_tq_osc4_palette_color_0() {
let results = parse_terminal_queries(b"\x1b]4;0;?\x07", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b]4;0;rgb:0000/0000/0000\x1b\\");
}
#[test]
fn parse_tq_osc4_palette_color_1() {
let results = parse_terminal_queries(b"\x1b]4;1;?\x07", (24, 80), (0, 0));
assert_eq!(results.len(), 1);
assert_eq!(results[0], "\x1b]4;1;rgb:8080/0000/0000\x1b\\");
}
#[test]
fn parse_tq_osc_mixed_with_csi() {
let results =
parse_terminal_queries(b"\x1b]11;?\x07\x1b[c\x1b]4;0;?\x07", (24, 80), (0, 0));
assert_eq!(results.len(), 3);
assert!(results[0].starts_with("\x1b]11;"));
assert!(results[1].starts_with("\x1b[?64;"));
assert!(results[2].starts_with("\x1b]4;0;"));
}
// ── build_search_results_msg ──
#[test]
fn search_results_empty() {
let msg = build_search_results_msg(42, &[]);
assert_eq!(msg[0], S2C_SEARCH_RESULTS);
assert_eq!(u16::from_le_bytes([msg[1], msg[2]]), 42);
assert_eq!(u16::from_le_bytes([msg[3], msg[4]]), 0);
assert_eq!(msg.len(), 5);
}
#[test]
fn search_results_single() {
let results = vec![SearchResultRow {
pty_id: 7,
score: 100,
primary_source: 1,
matched_sources: 3,
context: "hello".into(),
scroll_offset: Some(42),
}];
let msg = build_search_results_msg(1, &results);
assert_eq!(msg[0], S2C_SEARCH_RESULTS);
assert_eq!(u16::from_le_bytes([msg[3], msg[4]]), 1);
let pty_id = u16::from_le_bytes([msg[5], msg[6]]);
assert_eq!(pty_id, 7);
let score = u32::from_le_bytes([msg[7], msg[8], msg[9], msg[10]]);
assert_eq!(score, 100);
assert_eq!(msg[11], 1);
assert_eq!(msg[12], 3);
let scroll = u32::from_le_bytes([msg[13], msg[14], msg[15], msg[16]]);
assert_eq!(scroll, 42);
let ctx_len = u16::from_le_bytes([msg[17], msg[18]]) as usize;
assert_eq!(ctx_len, 5);
assert_eq!(&msg[19..19 + ctx_len], b"hello");
}
#[test]
fn search_results_none_scroll_offset() {
let results = vec![SearchResultRow {
pty_id: 1,
score: 0,
primary_source: 0,
matched_sources: 0,
context: String::new(),
scroll_offset: None,
}];
let msg = build_search_results_msg(0, &results);
let scroll = u32::from_le_bytes([msg[13], msg[14], msg[15], msg[16]]);
assert_eq!(scroll, u32::MAX);
}
// ── allocate_pty_id ──
#[test]
fn allocate_pty_id_empty_session() {
let mut sess = Session::new();
assert_eq!(sess.allocate_pty_id(0), Some(1));
}
#[test]
fn allocate_pty_id_rotates() {
let mut sess = Session::new();
// Sequential allocations return increasing IDs (not always 1).
assert_eq!(sess.allocate_pty_id(0), Some(1));
assert_eq!(sess.allocate_pty_id(0), Some(2));
assert_eq!(sess.allocate_pty_id(0), Some(3));
}
#[test]
fn allocate_pty_id_wraps_at_max() {
let mut sess = Session::new();
sess.next_pty_id = u16::MAX;
assert_eq!(sess.allocate_pty_id(0), Some(u16::MAX));
// Next allocation wraps to 1.
assert_eq!(sess.allocate_pty_id(0), Some(1));
}
// ── try_send_update ──
#[test]
fn try_send_no_change() {
let mut client = test_client();
let frame = sample_frame("x");
let now = Instant::now();
let outcome = try_send_update(&mut client, 1, frame, None, now, false);
assert!(matches!(outcome, SendOutcome::NoChange));
}
#[test]
fn try_send_sent() {
let (mut client, _rx) = test_client_with_capacity(8);
let frame = sample_frame("x");
let now = Instant::now();
let outcome = try_send_update(
&mut client,
1,
frame.clone(),
Some(vec![1, 2, 3]),
now,
true,
);
assert!(matches!(outcome, SendOutcome::Sent));
assert!(client.last_sent.contains_key(&1));
}
#[test]
fn try_send_backpressured_on_disconnect() {
let (mut client, rx) = test_client_with_capacity(0);
let frame = sample_frame("x");
let now = Instant::now();
// Drop the receiver to simulate a disconnected client.
drop(rx);
let outcome = try_send_update(
&mut client,
1,
frame.clone(),
Some(vec![1, 2, 3]),
now,
true,
);
assert!(matches!(outcome, SendOutcome::Backpressured));
assert!(
client.last_sent.contains_key(&1),
"last_sent should advance even on disconnect"
);
}
}