diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/mod.rs | 7 | ||||
| -rw-r--r-- | src/api/schema.rs | 18 | ||||
| -rw-r--r-- | src/api/schema/panes.rs | 55 | ||||
| -rw-r--r-- | src/api/schema/response.rs | 4 | ||||
| -rw-r--r-- | src/api/server.rs | 64 | ||||
| -rw-r--r-- | src/api/server/pane_graphics_stream.rs | 1178 |
6 files changed, 1324 insertions, 2 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index dad4cfd8..b0d7caed 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -7,6 +7,7 @@ mod subscriptions; mod wait; pub use event_hub::EventHub; +pub(crate) use server::cancel_inactive_pane_graphics_streams; pub use server::{start_server, start_server_with_capabilities, ServerHandle}; pub use status::{read_runtime_status_at, RuntimeStatus}; @@ -51,6 +52,12 @@ pub(crate) fn request_changes_ui(request: &Request) -> bool { | Method::PaneResize(_) | Method::PaneFocus(_) | Method::PaneRename(_) + | Method::PaneGraphicsSet(_) + | Method::PaneGraphicsClear(_) + | Method::PaneGraphicsStream(_) + | Method::PaneGraphicsStreamSet(_) + | Method::PaneGraphicsStreamOpen(_) + | Method::PaneGraphicsStreamClose(_) | Method::PaneReportAgent(_) | Method::PaneReportAgentSession(_) | Method::PaneReportMetadata(_) diff --git a/src/api/schema.rs b/src/api/schema.rs index 4fb3d841..4449b2d9 100644 --- a/src/api/schema.rs +++ b/src/api/schema.rs @@ -161,6 +161,24 @@ pub enum Method { PaneSendInput(PaneSendInputParams), #[serde(rename = "pane.read")] PaneRead(PaneReadParams), + #[serde(rename = "pane.graphics.set")] + PaneGraphicsSet(PaneGraphicsSetParams), + #[serde(rename = "pane.graphics.clear")] + PaneGraphicsClear(PaneGraphicsClearParams), + #[serde(rename = "pane.graphics.info")] + PaneGraphicsInfo(PaneTarget), + #[serde(rename = "pane.graphics.stream")] + #[schemars(skip)] + PaneGraphicsStream(PaneGraphicsStreamParams), + #[serde(skip)] + #[schemars(skip)] + PaneGraphicsStreamSet(PaneGraphicsSetParams), + #[serde(skip)] + #[schemars(skip)] + PaneGraphicsStreamOpen(PaneGraphicsStreamParams), + #[serde(skip)] + #[schemars(skip)] + PaneGraphicsStreamClose(PaneGraphicsStreamParams), #[serde(rename = "pane.report_agent")] PaneReportAgent(PaneReportAgentParams), #[serde(rename = "pane.report_agent_session")] diff --git a/src/api/schema/panes.rs b/src/api/schema/panes.rs index 2e37e734..097b6366 100644 --- a/src/api/schema/panes.rs +++ b/src/api/schema/panes.rs @@ -2,6 +2,9 @@ use std::collections::HashMap; use serde::{Deserialize, Serialize}; +pub(crate) const PANE_GRAPHICS_SET_MAX_BYTES: usize = 512 * 1024; +pub(crate) const PANE_GRAPHICS_STREAM_MAX_BYTES: usize = 16 * 1024 * 1024; + use super::agents::AgentSessionInfo; use super::common::{AgentStatus, PaneAgentState, ReadFormat, ReadSource, SplitDirection}; @@ -257,6 +260,58 @@ pub struct PaneReadParams { pub strip_ansi: bool, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum PaneGraphicsFormat { + Png, + Rgb, + Rgba, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] +pub struct PaneGraphicsSetParams { + pub pane_id: String, + #[serde(skip)] + #[schemars(skip)] + pub owner: String, + pub format: PaneGraphicsFormat, + pub image_width: u32, + pub image_height: u32, + #[serde(skip)] + pub data: Option<Vec<u8>>, + #[serde(default)] + pub data_base64: String, + #[serde(default)] + pub placement: PaneGraphicsPlacementParams, +} + +#[derive( + Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, Default, +)] +pub struct PaneGraphicsPlacementParams { + #[serde(default)] + pub viewport_col: i32, + #[serde(default)] + pub viewport_row: i32, + #[serde(default)] + pub grid_cols: u32, + #[serde(default)] + pub grid_rows: u32, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] +pub struct PaneGraphicsClearParams { + pub pane_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] +pub struct PaneGraphicsStreamParams { + pub pane_id: String, + #[serde(skip)] + #[schemars(skip)] + pub owner: String, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] pub struct PaneReportAgentParams { pub pane_id: String, diff --git a/src/api/schema/response.rs b/src/api/schema/response.rs index 18a2635d..129c8f91 100644 --- a/src/api/schema/response.rs +++ b/src/api/schema/response.rs @@ -152,6 +152,10 @@ pub enum ResponseResult { PaneRead { read: PaneReadResult, }, + PaneGraphicsInfo { + cell_width_px: u32, + cell_height_px: u32, + }, AgentExplain { explain: serde_json::Value, }, diff --git a/src/api/server.rs b/src/api/server.rs index 752893c6..aec7274d 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -22,6 +22,9 @@ use crate::ipc::{ socket_file_identity, LocalStream, LocalStreamRead, SocketFileIdentity, }; +mod pane_graphics_stream; +pub(crate) use pane_graphics_stream::cancel_inactive_streams as cancel_inactive_pane_graphics_streams; + const SOCKET_PERMISSION_MODE: u32 = 0o600; pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100); pub(super) const APP_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); @@ -155,14 +158,14 @@ fn handle_connection( let request = match serde_json::from_str::<Request>(line) { Ok(request) => request, - Err(err) => { + Err(request_error) => { write_json_line_allow_disconnect( &mut stream, &ErrorResponse { id: String::new(), error: ErrorBody { code: "invalid_request".into(), - message: format!("invalid request: {err}"), + message: format!("invalid request: {request_error}"), }, }, )?; @@ -176,6 +179,22 @@ fn handle_connection( crate::logging::api_request_started(&request_id, method, changes_ui); match request.method { + Method::PaneGraphicsStream(params) => { + let result = + pane_graphics_stream::serve(stream, request_id.clone(), params, api_tx, running); + match &result { + Ok(()) => crate::logging::api_request_completed( + &request_id, + method, + "stream_closed", + changes_ui, + ), + Err(err) => { + crate::logging::api_request_failed(&request_id, method, &err.to_string()) + } + } + result + } Method::EventsSubscribe(params) => { let result = stream_subscriptions( stream, @@ -374,6 +393,13 @@ fn api_method_name(method: &Method) -> &'static str { Method::PaneSendKeys(_) => "pane.send_keys", Method::PaneSendInput(_) => "pane.send_input", Method::PaneRead(_) => "pane.read", + Method::PaneGraphicsSet(_) => "pane.graphics.set", + Method::PaneGraphicsClear(_) => "pane.graphics.clear", + Method::PaneGraphicsInfo(_) => "pane.graphics.info", + Method::PaneGraphicsStream(_) => "pane.graphics.stream", + Method::PaneGraphicsStreamSet(_) => "pane.graphics.stream.set", + Method::PaneGraphicsStreamOpen(_) => "pane.graphics.stream.open", + Method::PaneGraphicsStreamClose(_) => "pane.graphics.stream.close", Method::PaneReportAgent(_) => "pane.report_agent", Method::PaneReportAgentSession(_) => "pane.report_agent_session", Method::PaneReportMetadata(_) => "pane.report_metadata", @@ -1223,3 +1249,37 @@ mod tests { server_thread.join().unwrap(); } } + +#[cfg(test)] +mod pane_graphics_request_tests { + use super::*; + use base64::Engine as _; + + #[test] + fn maximum_public_graphics_request_fits_initial_json_line() { + let request = Request { + id: "graphics-max".into(), + method: Method::PaneGraphicsSet(crate::api::schema::PaneGraphicsSetParams { + pane_id: "pane_1".into(), + owner: String::new(), + format: crate::api::schema::PaneGraphicsFormat::Png, + image_width: 1, + image_height: 1, + data_base64: base64::engine::general_purpose::STANDARD + .encode(vec![1_u8; crate::api::schema::PANE_GRAPHICS_SET_MAX_BYTES]), + data: None, + placement: crate::api::schema::PaneGraphicsPlacementParams::default(), + }), + }; + let encoded = serde_json::to_vec(&request).unwrap(); + + assert!(encoded.len() < MAX_INITIAL_REQUEST_BYTES); + } + + #[test] + fn duplicate_method_cannot_be_reinterpreted_as_graphics_stream() { + let encoded = r#"{"id":"duplicate","method":"ping","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#; + + assert!(serde_json::from_str::<Request>(encoded).is_err()); + } +} diff --git a/src/api/server/pane_graphics_stream.rs b/src/api/server/pane_graphics_stream.rs new file mode 100644 index 00000000..c6c58052 --- /dev/null +++ b/src/api/server/pane_graphics_stream.rs @@ -0,0 +1,1178 @@ +use std::collections::HashMap; +use std::io::{self, Read}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, OnceLock, Weak}; +use std::time::{Duration, Instant}; + +use interprocess::local_socket::traits::Stream as _; + +use crate::api::schema::{ + ErrorBody, ErrorResponse, Method, PaneGraphicsSetParams, PaneGraphicsStreamParams, Request, + ResponseResult, SuccessResponse, +}; +use crate::api::ApiRequestSender; +use crate::ipc::{is_connection_closed_error, LocalStream}; + +use super::{ + api_response_outcome, dispatch_to_app_with_timeout, write_json_line, + write_json_line_allow_disconnect, write_text_line_allow_disconnect, APP_RESPONSE_TIMEOUT, + CONNECTION_POLL_INTERVAL, +}; + +const MAX_STREAM_FRAME_HEADER_BYTES: usize = 64 * 1024; +const STREAM_FRAME_BODY_CHUNK_BYTES: usize = 64 * 1024; +const STREAM_FRAME_HEADER_IDLE_TIMEOUT: Duration = Duration::from_secs(5); +const STREAM_FRAME_HEADER_TIMEOUT: Duration = Duration::from_secs(30); +const STREAM_FRAME_BODY_IDLE_TIMEOUT: Duration = Duration::from_secs(5); +const STREAM_FRAME_BODY_TIMEOUT: Duration = Duration::from_secs(30); +const STREAM_FALLBACK_POLL_INTERVAL: Duration = Duration::from_millis(1); +const STREAM_FALLBACK_FAST_POLLS: u8 = 32; +static NEXT_PANE_GRAPHICS_STREAM_OWNER: AtomicU64 = AtomicU64::new(1); +static REGISTERED_STREAM_COUNT: AtomicUsize = AtomicUsize::new(0); +static REGISTERED_STREAMS: OnceLock<Mutex<HashMap<String, Weak<AtomicBool>>>> = OnceLock::new(); + +#[derive(serde::Deserialize)] +struct FrameHeader { + format: crate::api::schema::PaneGraphicsFormat, + image_width: u32, + image_height: u32, + data_length: usize, + #[serde(default)] + placement: crate::api::schema::PaneGraphicsPlacementParams, +} + +#[derive(Clone, Copy)] +struct ReadTimeouts { + header_idle: Duration, + header_total: Duration, + body_idle: Duration, + body_total: Duration, +} + +const READ_TIMEOUTS: ReadTimeouts = ReadTimeouts { + header_idle: STREAM_FRAME_HEADER_IDLE_TIMEOUT, + header_total: STREAM_FRAME_HEADER_TIMEOUT, + body_idle: STREAM_FRAME_BODY_IDLE_TIMEOUT, + body_total: STREAM_FRAME_BODY_TIMEOUT, +}; + +pub(super) fn serve( + stream: LocalStream, + request_id: String, + params: PaneGraphicsStreamParams, + api_tx: &ApiRequestSender, + running: &Arc<AtomicBool>, +) -> std::io::Result<()> { + serve_with_open_timeout( + stream, + request_id, + params, + api_tx, + running, + APP_RESPONSE_TIMEOUT, + ) +} + +fn serve_with_open_timeout( + stream: LocalStream, + request_id: String, + params: PaneGraphicsStreamParams, + api_tx: &ApiRequestSender, + running: &Arc<AtomicBool>, + open_timeout: Duration, +) -> std::io::Result<()> { + serve_with_timeouts( + stream, + request_id, + params, + api_tx, + running, + open_timeout, + READ_TIMEOUTS, + ) +} + +fn serve_with_timeouts( + mut stream: LocalStream, + request_id: String, + mut params: PaneGraphicsStreamParams, + api_tx: &ApiRequestSender, + running: &Arc<AtomicBool>, + open_timeout: Duration, + read_timeouts: ReadTimeouts, +) -> std::io::Result<()> { + let pane_id = params.pane_id.clone(); + let owner = next_owner(); + params.owner = owner.clone(); + let open_response = dispatch_to_app_with_timeout( + Request { + id: request_id.clone(), + method: Method::PaneGraphicsStreamOpen(params), + }, + api_tx, + Some(open_timeout), + ); + if api_response_outcome(&open_response) != "ok" { + let write_result = write_text_line_allow_disconnect(&mut stream, &open_response); + clear_layer(&pane_id, &owner, api_tx); + write_result?; + return Ok(()); + } + + if let Err(err) = write_json_line( + &mut stream, + &SuccessResponse { + id: request_id.clone(), + result: ResponseResult::Ok {}, + }, + ) { + clear_layer(&pane_id, &owner, api_tx); + if is_connection_closed_error(&err) { + return Ok(()); + } + return Err(err); + } + + let stream_active = Arc::new(AtomicBool::new(true)); + register_stream(&owner, &stream_active); + let result = serve_frames( + &mut stream, + &request_id, + &owner, + &pane_id, + api_tx, + running, + &stream_active, + read_timeouts, + ); + unregister_stream(&owner); + clear_layer(&pane_id, &owner, api_tx); + result +} + +fn serve_frames( + stream: &mut LocalStream, + request_id: &str, + owner: &str, + pane_id: &str, + api_tx: &ApiRequestSender, + running: &Arc<AtomicBool>, + stream_active: &Arc<AtomicBool>, + timeouts: ReadTimeouts, +) -> std::io::Result<()> { + let mut frame_seq = 0_u64; + while stream_is_running(running, stream_active) { + let Some(header_line) = read_line( + stream, + running, + stream_active, + MAX_STREAM_FRAME_HEADER_BYTES, + timeouts.header_idle, + timeouts.header_total, + )? + else { + return Ok(()); + }; + let header_line = header_line.trim(); + if header_line.is_empty() { + continue; + } + let header = match serde_json::from_str::<FrameHeader>(header_line) { + Ok(header) => header, + Err(err) => { + write_json_line_allow_disconnect( + stream, + &ErrorResponse { + id: request_id.to_string(), + error: ErrorBody { + code: "invalid_frame".into(), + message: format!("invalid frame header: {err}"), + }, + }, + )?; + return Ok(()); + } + }; + if header.data_length == 0 { + write_json_line_allow_disconnect( + stream, + &ErrorResponse { + id: request_id.to_string(), + error: ErrorBody { + code: "invalid_frame".into(), + message: "frame data_length must be greater than zero".into(), + }, + }, + )?; + return Ok(()); + } + if header.data_length > crate::api::schema::PANE_GRAPHICS_STREAM_MAX_BYTES { + write_json_line_allow_disconnect( + stream, + &ErrorResponse { + id: request_id.to_string(), + error: ErrorBody { + code: "image_too_large".into(), + message: "frame data is too large".into(), + }, + }, + )?; + return Ok(()); + } + + let Some(data) = read_exact( + stream, + header.data_length, + running, + stream_active, + timeouts.body_idle, + timeouts.body_total, + )? + else { + return Ok(()); + }; + + frame_seq = frame_seq.saturating_add(1); + let frame_id = format!("{request_id}:frame:{frame_seq}"); + let response = dispatch_to_app_with_timeout( + Request { + id: frame_id, + method: Method::PaneGraphicsStreamSet(PaneGraphicsSetParams { + pane_id: pane_id.to_string(), + owner: owner.to_string(), + format: header.format, + image_width: header.image_width, + image_height: header.image_height, + data: Some(data), + data_base64: String::new(), + placement: header.placement, + }), + }, + api_tx, + Some(APP_RESPONSE_TIMEOUT), + ); + if api_response_outcome(&response) != "ok" { + write_text_line_allow_disconnect(stream, &response)?; + return Ok(()); + } + } + + Ok(()) +} + +fn stream_registry() -> &'static Mutex<HashMap<String, Weak<AtomicBool>>> { + REGISTERED_STREAMS.get_or_init(|| Mutex::new(HashMap::new())) +} + +fn register_stream(owner: &str, active: &Arc<AtomicBool>) { + let Ok(mut streams) = stream_registry().lock() else { + active.store(false, Ordering::Release); + return; + }; + if streams + .insert(owner.to_string(), Arc::downgrade(active)) + .is_none() + { + REGISTERED_STREAM_COUNT.fetch_add(1, Ordering::Release); + } +} + +fn unregister_stream(owner: &str) { + let Ok(mut streams) = stream_registry().lock() else { + return; + }; + if streams.remove(owner).is_some() { + REGISTERED_STREAM_COUNT.fetch_sub(1, Ordering::Release); + } +} + +pub(crate) fn cancel_inactive_streams(mut is_active: impl FnMut(&str) -> bool) { + if REGISTERED_STREAM_COUNT.load(Ordering::Acquire) == 0 { + return; + } + let Ok(mut streams) = stream_registry().lock() else { + return; + }; + let before = streams.len(); + streams.retain(|owner, active| { + let keep = is_active(owner); + if !keep { + if let Some(active) = active.upgrade() { + active.store(false, Ordering::Release); + } + } + keep + }); + REGISTERED_STREAM_COUNT.fetch_sub(before.saturating_sub(streams.len()), Ordering::Release); +} + +fn stream_is_running(running: &AtomicBool, stream_active: &AtomicBool) -> bool { + running.load(Ordering::Relaxed) && stream_active.load(Ordering::Acquire) +} + +fn next_owner() -> String { + let id = NEXT_PANE_GRAPHICS_STREAM_OWNER.fetch_add(1, Ordering::Relaxed); + format!("pane.graphics.stream:{}:{id}", std::process::id()) +} + +fn clear_layer(pane_id: &str, owner: &str, api_tx: &ApiRequestSender) { + let _response = dispatch_to_app_with_timeout( + Request { + id: format!("pane.graphics.stream.clear:{pane_id}"), + method: Method::PaneGraphicsStreamClose(PaneGraphicsStreamParams { + pane_id: pane_id.to_string(), + owner: owner.to_string(), + }), + }, + api_tx, + Some(APP_RESPONSE_TIMEOUT), + ); +} + +fn read_line( + stream: &mut LocalStream, + running: &Arc<AtomicBool>, + stream_active: &Arc<AtomicBool>, + max_bytes: usize, + idle_timeout: Duration, + total_timeout: Duration, +) -> std::io::Result<Option<String>> { + with_timed_reads(stream, |stream, mut wait| { + let mut bytes = Vec::new(); + let mut byte = [0_u8; 1]; + let mut total_deadline = None; + let mut idle_deadline = None; + + loop { + if !stream_is_running(running, stream_active) { + return Ok(None); + } + ensure_before_deadlines( + idle_deadline, + total_deadline, + "timed out reading stream frame header", + )?; + match stream.read(&mut byte) { + Ok(0) => return Ok(None), + Ok(_) => { + wait.on_progress(); + let now = Instant::now(); + let total_deadline_at = + *total_deadline.get_or_insert_with(|| now + total_timeout); + idle_deadline = Some(now + idle_timeout); + if now >= total_deadline_at { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "timed out reading stream frame header", + )); + } + bytes.push(byte[0]); + if byte[0] == b'\n' { + return String::from_utf8(bytes) + .map(Some) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err)); + } + if bytes.len() > max_bytes { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "stream frame header is too large", + )); + } + } + Err(err) if read_should_retry(&err) => { + wait.after_retry(idle_deadline, total_deadline); + } + Err(err) if is_connection_closed_error(&err) => return Ok(None), + Err(err) => return Err(err), + } + } + }) +} + +fn read_exact( + stream: &mut LocalStream, + len: usize, + running: &Arc<AtomicBool>, + stream_active: &Arc<AtomicBool>, + idle_timeout: Duration, + total_timeout: Duration, +) -> std::io::Result<Option<Vec<u8>>> { + with_timed_reads(stream, |stream, mut wait| { + let mut data = Vec::new(); + let mut chunk = vec![0_u8; STREAM_FRAME_BODY_CHUNK_BYTES.min(len)]; + let total_deadline = Instant::now() + total_timeout; + let mut idle_deadline = Instant::now() + idle_timeout; + + while data.len() < len { + if !stream_is_running(running, stream_active) { + return Ok(None); + } + ensure_before_deadlines( + Some(idle_deadline), + Some(total_deadline), + "timed out reading stream frame body", + )?; + let remaining = len - data.len(); + let read_len = remaining.min(chunk.len()); + match stream.read(&mut chunk[..read_len]) { + Ok(0) if data.is_empty() => return Ok(None), + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "stream ended mid-frame", + )); + } + Ok(n) => { + wait.on_progress(); + let now = Instant::now(); + if now >= total_deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "timed out reading stream frame body", + )); + } + data.extend_from_slice(&chunk[..n]); + idle_deadline = now + idle_timeout; + } + Err(err) if read_should_retry(&err) => { + wait.after_retry(Some(idle_deadline), Some(total_deadline)); + } + Err(err) if is_connection_closed_error(&err) && data.is_empty() => return Ok(None), + Err(err) => return Err(err), + } + } + + Ok(Some(data)) + }) +} + +#[derive(Clone, Copy)] +enum ReadWait { + SocketTimeout, + Poll(PollBackoff), +} + +impl ReadWait { + fn after_retry(&mut self, idle_deadline: Option<Instant>, total_deadline: Option<Instant>) { + if let Self::Poll(backoff) = self { + sleep_until_poll(idle_deadline, total_deadline, backoff.interval); + backoff.advance(); + } + } + + fn on_progress(&mut self) { + if let Self::Poll(backoff) = self { + backoff.reset(); + } + } +} + +#[derive(Clone, Copy)] +struct PollBackoff { + interval: Duration, + fast_polls_remaining: u8, +} + +impl PollBackoff { + fn new() -> Self { + Self { + interval: STREAM_FALLBACK_POLL_INTERVAL, + fast_polls_remaining: STREAM_FALLBACK_FAST_POLLS, + } + } + + fn advance(&mut self) { + if self.fast_polls_remaining > 0 { + self.fast_polls_remaining -= 1; + return; + } + self.interval = (self.interval * 2).min(CONNECTION_POLL_INTERVAL); + } + + fn reset(&mut self) { + *self = Self::new(); + } +} + +fn with_timed_reads<T>( + stream: &mut LocalStream, + read: impl FnOnce(&mut LocalStream, ReadWait) -> std::io::Result<T>, +) -> std::io::Result<T> { + match stream.set_recv_timeout(Some(CONNECTION_POLL_INTERVAL)) { + Ok(()) => { + let result = read(stream, ReadWait::SocketTimeout); + merge_read_reset_result(result, stream.set_recv_timeout(None)) + } + Err(err) if err.kind() == io::ErrorKind::Unsupported => { + stream.set_nonblocking(true)?; + let result = read(stream, ReadWait::Poll(PollBackoff::new())); + merge_read_reset_result(result, stream.set_nonblocking(false)) + } + Err(err) => Err(err), + } +} + +fn merge_read_reset_result<T>( + result: std::io::Result<T>, + reset_result: std::io::Result<()>, +) -> std::io::Result<T> { + match (result, reset_result) { + (Ok(value), Ok(())) => Ok(value), + (Ok(_), Err(err)) => Err(err), + (Err(err), _) => Err(err), + } +} + +fn ensure_before_deadlines( + idle_deadline: Option<Instant>, + total_deadline: Option<Instant>, + message: &str, +) -> std::io::Result<()> { + let now = Instant::now(); + if idle_deadline.is_some_and(|deadline| now >= deadline) + || total_deadline.is_some_and(|deadline| now >= deadline) + { + return Err(io::Error::new(io::ErrorKind::TimedOut, message)); + } + Ok(()) +} + +fn sleep_until_poll( + idle_deadline: Option<Instant>, + total_deadline: Option<Instant>, + poll_interval: Duration, +) { + let now = Instant::now(); + let until_deadline = [idle_deadline, total_deadline] + .into_iter() + .flatten() + .filter_map(|deadline| deadline.checked_duration_since(now)) + .min() + .unwrap_or(poll_interval); + std::thread::sleep(poll_interval.min(until_deadline)); +} + +fn read_should_retry(err: &io::Error) -> bool { + matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut | io::ErrorKind::Interrupted + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::api::schema::{ErrorResponse, Method, ResponseResult, SuccessResponse}; + use crate::api::{ApiRequestMessage, EventHub}; + use crate::ipc::LocalStream; + use interprocess::local_socket::traits::Listener as _; + use std::io::{BufRead, BufReader, Write}; + use std::path::PathBuf; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + use std::time::{Duration, Instant}; + use tokio::sync::mpsc; + + fn local_stream_pair(name: &str) -> (LocalStream, LocalStream, PathBuf) { + let unique = format!( + "herdr-{name}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let path = std::env::temp_dir().join(unique); + let listener = crate::ipc::bind_local_listener(&path).unwrap(); + let client = crate::ipc::connect_local_stream(&path).unwrap(); + let server = listener.accept().unwrap(); + (client, server, path) + } + + fn read_response_line(stream: &mut LocalStream) -> String { + let mut reader = BufReader::new(stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + line + } + + fn assert_server_stream_owner(owner: &str) { + assert!(owner.starts_with("pane.graphics.stream:")); + } + + #[cfg(unix)] + #[test] + fn pane_graphics_stream_dispatches_binary_frames() { + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let (mut client, server, _path) = local_stream_pair("api-pane-graphics-stream"); + client + .write_all(br#"{"id":"stream_1","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let event_hub = EventHub::default(); + let server_thread = std::thread::spawn(move || { + super::super::handle_connection(server, &api_tx, &event_hub, &server_running, None) + }); + + let open = api_rx.blocking_recv().unwrap(); + let stream_owner = match &open.request.method { + Method::PaneGraphicsStreamOpen(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_server_stream_owner(¶ms.owner); + assert_ne!(params.owner, "stream_1"); + params.owner.clone() + } + other => panic!("unexpected open request: {other:?}"), + }; + open.respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: open.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + let ack: SuccessResponse = serde_json::from_str(&read_response_line(&mut client)).unwrap(); + assert_eq!(ack.id, "stream_1"); + assert_eq!(ack.result, ResponseResult::Ok {}); + + client + .write_all( + br#"{"format":"png","image_width":2,"image_height":1,"data_length":4,"placement":{"grid_cols":10,"grid_rows":5}}"#, + ) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.write_all(&[1_u8, 2, 3, 4]).unwrap(); + client.flush().unwrap(); + + let msg = api_rx.blocking_recv().unwrap(); + match &msg.request.method { + Method::PaneGraphicsStreamSet(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_eq!(params.owner, stream_owner); + assert_eq!(params.image_width, 2); + assert_eq!(params.image_height, 1); + assert_eq!(params.data.as_deref(), Some(&[1_u8, 2, 3, 4][..])); + assert!(params.data_base64.is_empty()); + assert_eq!(params.placement.grid_cols, 10); + assert_eq!(params.placement.grid_rows, 5); + } + other => panic!("unexpected request: {other:?}"), + } + msg.respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: msg.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + drop(client); + running.store(false, Ordering::Relaxed); + let clear = api_rx.blocking_recv().unwrap(); + match &clear.request.method { + Method::PaneGraphicsStreamClose(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_eq!(params.owner, stream_owner); + } + other => panic!("unexpected clear request: {other:?}"), + } + clear + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: clear.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + assert!(server_thread.join().unwrap().is_ok()); + } + + #[cfg(unix)] + #[test] + fn pane_graphics_stream_reports_open_errors_before_ack() { + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let (mut client, server, _path) = local_stream_pair("api-pane-graphics-stream-error"); + client + .write_all(br#"{"id":"stream_2","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let event_hub = EventHub::default(); + let server_thread = std::thread::spawn(move || { + super::super::handle_connection(server, &api_tx, &event_hub, &server_running, None) + }); + + let open = api_rx.blocking_recv().unwrap(); + let stream_owner = match &open.request.method { + Method::PaneGraphicsStreamOpen(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_server_stream_owner(¶ms.owner); + assert_ne!(params.owner, "stream_2"); + params.owner.clone() + } + other => panic!("unexpected open request: {other:?}"), + }; + open.respond_to + .send(super::super::error_response_json( + open.request.id, + "feature_disabled", + "pane graphics require experimental.kitty_graphics".into(), + )) + .unwrap(); + + let response: ErrorResponse = + serde_json::from_str(&read_response_line(&mut client)).unwrap(); + assert_eq!(response.id, "stream_2"); + assert_eq!(response.error.code, "feature_disabled"); + + let close = api_rx.blocking_recv().unwrap(); + match &close.request.method { + Method::PaneGraphicsStreamClose(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_eq!(params.owner, stream_owner); + } + other => panic!("unexpected close request: {other:?}"), + } + close + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: close.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + drop(client); + running.store(false, Ordering::Relaxed); + assert!(server_thread.join().unwrap().is_ok()); + } + + #[cfg(unix)] + #[test] + fn pane_graphics_stream_closes_claim_after_open_timeout() { + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let (mut client, server, _path) = local_stream_pair("api-pane-graphics-stream-timeout"); + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let server_thread = std::thread::spawn(move || { + serve_with_open_timeout( + server, + "stream_timeout".into(), + PaneGraphicsStreamParams { + pane_id: "pane_1".into(), + owner: String::new(), + }, + &api_tx, + &server_running, + Duration::from_millis(10), + ) + }); + + let open = api_rx.blocking_recv().unwrap(); + let stream_owner = match &open.request.method { + Method::PaneGraphicsStreamOpen(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_server_stream_owner(¶ms.owner); + assert_ne!(params.owner, "stream_timeout"); + params.owner.clone() + } + other => panic!("unexpected open request: {other:?}"), + }; + + let response: ErrorResponse = + serde_json::from_str(&read_response_line(&mut client)).unwrap(); + assert_eq!(response.id, "stream_timeout"); + assert_eq!(response.error.code, "server_unavailable"); + assert!(response.error.message.contains("timed out")); + + let close = api_rx.blocking_recv().unwrap(); + match &close.request.method { + Method::PaneGraphicsStreamClose(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_eq!(params.owner, stream_owner); + } + other => panic!("unexpected close request: {other:?}"), + } + close + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: close.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + drop(open); + drop(client); + running.store(false, Ordering::Relaxed); + assert!(server_thread.join().unwrap().is_ok()); + } + + #[cfg(unix)] + #[test] + fn pane_graphics_stream_closes_claim_when_client_disconnects_before_ack() { + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let (mut client, server, _path) = + local_stream_pair("api-pane-graphics-stream-ack-disconnect"); + client + .write_all(br#"{"id":"stream_3","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let event_hub = EventHub::default(); + let server_thread = std::thread::spawn(move || { + super::super::handle_connection(server, &api_tx, &event_hub, &server_running, None) + }); + + let open = api_rx.blocking_recv().unwrap(); + let stream_owner = match &open.request.method { + Method::PaneGraphicsStreamOpen(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_server_stream_owner(¶ms.owner); + assert_ne!(params.owner, "stream_3"); + params.owner.clone() + } + other => panic!("unexpected open request: {other:?}"), + }; + + drop(client); + open.respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: open.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + let (close_tx, close_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + close_tx.send(api_rx.blocking_recv()).unwrap(); + }); + let close = close_rx + .recv_timeout(Duration::from_secs(2)) + .unwrap() + .unwrap(); + match &close.request.method { + Method::PaneGraphicsStreamClose(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_eq!(params.owner, stream_owner); + } + other => panic!("unexpected close request: {other:?}"), + } + close + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: close.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + running.store(false, Ordering::Relaxed); + assert!(server_thread.join().unwrap().is_ok()); + } + + #[test] + fn idle_graphics_stream_waits_for_header_without_timing_out() { + let (_client, mut server, _path) = local_stream_pair("graphics-idle-header"); + let running = Arc::new(AtomicBool::new(true)); + let active = Arc::new(AtomicBool::new(true)); + let stop = Arc::clone(&running); + let stopper = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(30)); + stop.store(false, Ordering::Relaxed); + }); + + let result = read_line( + &mut server, + &running, + &active, + MAX_STREAM_FRAME_HEADER_BYTES, + Duration::from_millis(10), + Duration::from_millis(20), + ) + .unwrap(); + + assert!(result.is_none()); + stopper.join().unwrap(); + } + + #[test] + fn inactive_owner_cancels_idle_stream_and_dispatches_close() { + let (mut client, server, _path) = local_stream_pair("graphics-owner-cancel"); + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let server_thread = std::thread::spawn(move || { + serve_with_timeouts( + server, + "stream-cancel".into(), + PaneGraphicsStreamParams { + pane_id: "pane_1".into(), + owner: String::new(), + }, + &api_tx, + &server_running, + Duration::from_secs(1), + READ_TIMEOUTS, + ) + }); + + let open = api_rx.blocking_recv().unwrap(); + let owner = match &open.request.method { + Method::PaneGraphicsStreamOpen(params) => params.owner.clone(), + other => panic!("unexpected open request: {other:?}"), + }; + open.respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: open.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + let ack: SuccessResponse = serde_json::from_str(&read_response_line(&mut client)).unwrap(); + assert_eq!(ack.id, "stream-cancel"); + + cancel_inactive_streams(|registered| registered != owner); + + let (close_tx, close_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || close_tx.send(api_rx.blocking_recv()).unwrap()); + let close = close_rx + .recv_timeout(Duration::from_secs(1)) + .expect("canceled idle stream should dispatch a close") + .expect("API request channel should remain open"); + match &close.request.method { + Method::PaneGraphicsStreamClose(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_eq!(params.owner, owner); + } + other => panic!("unexpected close request: {other:?}"), + } + close + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: close.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + drop(client); + running.store(false, Ordering::Relaxed); + assert!(server_thread.join().unwrap().is_ok()); + } + + #[test] + fn fallback_poll_backoff_preserves_fast_window_then_reaches_poll_ceiling() { + let mut backoff = PollBackoff::new(); + for _ in 0..STREAM_FALLBACK_FAST_POLLS { + backoff.advance(); + assert_eq!(backoff.interval, STREAM_FALLBACK_POLL_INTERVAL); + } + + backoff.advance(); + assert_eq!(backoff.interval, Duration::from_millis(2)); + for _ in 0..6 { + backoff.advance(); + } + assert_eq!(backoff.interval, CONNECTION_POLL_INTERVAL); + + backoff.reset(); + assert_eq!(backoff.interval, STREAM_FALLBACK_POLL_INTERVAL); + assert_eq!(backoff.fast_polls_remaining, STREAM_FALLBACK_FAST_POLLS); + } + + #[test] + fn partial_graphics_header_times_out_after_first_byte() { + let (mut client, mut server, _path) = local_stream_pair("graphics-partial-header"); + client.write_all(b"{").unwrap(); + client.flush().unwrap(); + let running = Arc::new(AtomicBool::new(true)); + let active = Arc::new(AtomicBool::new(true)); + + let error = read_line( + &mut server, + &running, + &active, + MAX_STREAM_FRAME_HEADER_BYTES, + Duration::from_millis(20), + Duration::from_millis(100), + ) + .unwrap_err(); + + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + } + + #[test] + fn trickled_graphics_body_obeys_absolute_deadline() { + let (mut client, mut server, _path) = local_stream_pair("graphics-trickle-body"); + client.write_all(&[1_u8]).unwrap(); + client.flush().unwrap(); + let running = Arc::new(AtomicBool::new(true)); + let active = Arc::new(AtomicBool::new(true)); + let writer_running = Arc::clone(&running); + let writer = std::thread::spawn(move || { + while writer_running.load(Ordering::Relaxed) { + if client.write_all(&[1_u8]).is_err() { + break; + } + let _ = client.flush(); + std::thread::sleep(Duration::from_millis(5)); + } + }); + + let started = Instant::now(); + let error = read_exact( + &mut server, + 1024, + &running, + &active, + Duration::from_millis(20), + Duration::from_millis(60), + ) + .unwrap_err(); + + running.store(false, Ordering::Relaxed); + writer.join().unwrap(); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + assert!(started.elapsed() >= Duration::from_millis(50)); + assert!(started.elapsed() < Duration::from_millis(500)); + } + + #[test] + fn timed_out_header_dispatches_owner_scoped_stream_close() { + let (mut client, server, _path) = local_stream_pair("graphics-timeout-close"); + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let server_thread = std::thread::spawn(move || { + serve_with_timeouts( + server, + "stream-timeout".into(), + PaneGraphicsStreamParams { + pane_id: "pane_1".into(), + owner: String::new(), + }, + &api_tx, + &server_running, + Duration::from_secs(1), + ReadTimeouts { + header_idle: Duration::from_millis(20), + header_total: Duration::from_millis(100), + body_idle: Duration::from_millis(20), + body_total: Duration::from_millis(100), + }, + ) + }); + + let open = api_rx.blocking_recv().unwrap(); + let owner = match &open.request.method { + Method::PaneGraphicsStreamOpen(params) => params.owner.clone(), + other => panic!("unexpected open request: {other:?}"), + }; + open.respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: open.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + let ack: SuccessResponse = serde_json::from_str(&read_response_line(&mut client)).unwrap(); + assert_eq!(ack.id, "stream-timeout"); + client.write_all(b"{").unwrap(); + client.flush().unwrap(); + + let (close_tx, close_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || close_tx.send(api_rx.blocking_recv()).unwrap()); + let close = close_rx + .recv_timeout(Duration::from_secs(1)) + .unwrap() + .unwrap(); + match &close.request.method { + Method::PaneGraphicsStreamClose(params) => { + assert_eq!(params.pane_id, "pane_1"); + assert_eq!(params.owner, owner); + } + other => panic!("unexpected close request: {other:?}"), + } + close + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: close.request.id, + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + let error = server_thread.join().unwrap().unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + } + + #[test] + fn oversized_stream_frame_is_rejected_before_body_or_app_dispatch() { + let (mut client, mut server, _path) = local_stream_pair("graphics-oversized-frame"); + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let stream_active = Arc::new(AtomicBool::new(true)); + let server_thread = std::thread::spawn(move || { + serve_frames( + &mut server, + "stream-oversized", + "owner-1", + "pane_1", + &api_tx, + &server_running, + &stream_active, + READ_TIMEOUTS, + ) + }); + let header = serde_json::json!({ + "format": "png", + "image_width": 1, + "image_height": 1, + "data_length": crate::api::schema::PANE_GRAPHICS_STREAM_MAX_BYTES + 1, + }); + client.write_all(format!("{header}\n").as_bytes()).unwrap(); + client.flush().unwrap(); + + let response: ErrorResponse = + serde_json::from_str(&read_response_line(&mut client)).unwrap(); + assert_eq!(response.error.code, "image_too_large"); + assert!(server_thread.join().unwrap().is_ok()); + assert!(api_rx.try_recv().is_err()); + } +} |