diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/mod.rs | 8 | ||||
| -rw-r--r-- | src/api/schema.rs | 21 | ||||
| -rw-r--r-- | src/api/schema/panes.rs | 86 | ||||
| -rw-r--r-- | src/api/schema/response.rs | 27 | ||||
| -rw-r--r-- | src/api/server.rs | 220 | ||||
| -rw-r--r-- | src/api/server/pane_graphics_stream.rs | 1114 |
6 files changed, 117 insertions, 1359 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index bd2983e4..86f90da8 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -64,13 +64,6 @@ pub(crate) fn request_changes_ui(request: &Request) -> bool { | Method::PaneFocus(_) | Method::PaneInputSet(_) | Method::PaneRename(_) - | Method::PaneGraphicsSet(_) - | Method::PaneGraphicsClear(_) - | Method::PaneGraphicsStream(_) - | Method::PaneGraphicsStreamSet(_) - | Method::PaneGraphicsStreamDirect(_) - | Method::PaneGraphicsStreamOpen(_) - | Method::PaneGraphicsStreamClose(_) | Method::PaneReportAgent(_) | Method::PaneReportAgentSession(_) | Method::PaneReportMetadata(_) @@ -91,7 +84,6 @@ pub struct ApiRequestMessage { pub request: Request, pub respond_to: std::sync::mpsc::Sender<String>, pub response_write_complete: Option<std::sync::mpsc::Receiver<()>>, - pub stream_active: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>, } pub type ApiRequestSender = mpsc::UnboundedSender<ApiRequestMessage>; diff --git a/src/api/schema.rs b/src/api/schema.rs index d7e1a950..439ed331 100644 --- a/src/api/schema.rs +++ b/src/api/schema.rs @@ -201,27 +201,6 @@ 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)] - PaneGraphicsStreamDirect(PaneGraphicsDirectParams), - #[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 67d5185e..dfef31f8 100644 --- a/src/api/schema/panes.rs +++ b/src/api/schema/panes.rs @@ -2,14 +2,6 @@ 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; -pub(crate) const PANE_GRAPHICS_DIRECT_FILE_MAX_BYTES: usize = 400 * 1024 * 1024; -pub(crate) const PANE_GRAPHICS_MAX_LAYERS_PER_PANE: usize = 16; -pub(crate) const PANE_GRAPHICS_MAX_LAYERS_TOTAL: usize = 64; -pub(crate) const PANE_GRAPHICS_MAX_INLINE_BYTES_TOTAL: usize = 64 * 1024 * 1024; -pub(crate) const PANE_GRAPHICS_PRIMARY_LAYER_ID: &str = "primary"; - use super::agents::AgentSessionInfo; use super::common::{AgentStatus, PaneAgentState, ReadFormat, ReadSource, SplitDirection}; @@ -366,84 +358,6 @@ pub struct PaneReadParams { pub(crate) intent: super::common::ReadIntent, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] -#[serde(rename_all = "snake_case")] -pub enum PaneGraphicsFormat { - Png, - Rgb, - Rgba, - Bgra, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] -pub struct PaneGraphicsSetParams { - pub pane_id: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub layer_id: Option<String>, - #[serde(default)] - pub z_index: i32, - #[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, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub layer_id: Option<String>, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct PaneGraphicsDirectParams { - pub pane_id: String, - pub layer_id: Option<String>, - pub z_index: i32, - pub owner: String, - pub image_width: u32, - pub image_height: u32, - pub format: PaneGraphicsFormat, - pub path: String, - pub sequence: u64, - pub revision: u64, - pub placement: PaneGraphicsPlacementParams, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] -pub struct PaneGraphicsStreamParams { - pub pane_id: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub layer_id: Option<String>, - #[serde(default)] - pub z_index: i32, - #[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 5cc66bdc..e92e4952 100644 --- a/src/api/schema/response.rs +++ b/src/api/schema/response.rs @@ -181,33 +181,6 @@ pub enum ResponseResult { #[serde(default, skip_serializing_if = "Option::is_none")] current_global: Option<u64>, }, - PaneGraphicsFrameAck { - sequence: u64, - revision: u64, - }, - PaneGraphicsInfo { - cell_width_px: u32, - cell_height_px: u32, - /// True only when this pane is on the currently rendered terminal surface. - pane_visible: bool, - #[serde(default, skip_serializing_if = "Option::is_none")] - file_frame_directory: Option<String>, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - file_frame_formats: Vec<String>, - #[serde(default, skip_serializing_if = "Option::is_none")] - file_frame_max_bytes: Option<usize>, - #[serde(default, skip_serializing_if = "Option::is_none")] - file_frame_direct_max_bytes: Option<usize>, - /// Accepts damage metadata while still consuming a complete canonical file. - #[serde(default)] - file_frame_damage: bool, - #[serde(default)] - max_layers_per_pane: usize, - #[serde(default)] - pixel_mouse: bool, - #[serde(default, skip_serializing_if = "Option::is_none")] - file_frame_transport: Option<String>, - }, AgentExplain { explain: serde_json::Value, }, diff --git a/src/api/server.rs b/src/api/server.rs index edb778f7..30d92be4 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -22,8 +22,6 @@ use crate::ipc::{ socket_file_identity, LocalStream, LocalStreamRead, SocketFileIdentity, }; -mod pane_graphics_stream; - #[cfg(test)] mod subscription_socket_tests; @@ -162,6 +160,30 @@ fn start_server_inner( }) } +fn retired_pane_graphics_method_error(line: &str, id: &str) -> Option<ErrorResponse> { + #[derive(serde::Deserialize)] + struct RequestMethod { + method: String, + } + + let envelope = serde_json::from_str::<RequestMethod>(line).ok()?; + let method = envelope.method.as_str(); + if !matches!( + method, + "pane.graphics.info" | "pane.graphics.set" | "pane.graphics.clear" | "pane.graphics.stream" + ) { + return None; + } + + Some(ErrorResponse { + id: id.into(), + error: ErrorBody { + code: "unknown_method".into(), + message: format!("unknown method: {method}"), + }, + }) +} + fn prepare_socket_path(path: &Path) -> std::io::Result<()> { crate::ipc::prepare_socket_path(path, |path| { format!( @@ -233,16 +255,15 @@ fn handle_connection_with_stop( } else { String::new() }; - write_json_line_allow_disconnect( - &mut stream, - &ErrorResponse { + let response = + retired_pane_graphics_method_error(line, &id).unwrap_or_else(|| ErrorResponse { id, error: ErrorBody { code: "invalid_request".into(), message: format!("invalid request: {request_error}"), }, - }, - )?; + }); + write_json_line_allow_disconnect(&mut stream, &response)?; return Ok(()); } }; @@ -296,22 +317,6 @@ fn handle_connection_with_stop( } Ok(()) } - 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, @@ -478,7 +483,7 @@ fn handle_request( ); } - dispatch_to_app(request, api_tx, None, response_write_complete, None, None) + dispatch_to_app(request, api_tx, None, response_write_complete, None) } pub(crate) fn api_method_name(method: &Method) -> &'static str { @@ -561,14 +566,6 @@ pub(crate) 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::PaneGraphicsStreamDirect(_) => "pane.graphics.stream.direct", - 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", @@ -928,7 +925,7 @@ pub(super) fn dispatch_to_app_with_timeout( api_tx: &ApiRequestSender, timeout: Option<Duration>, ) -> String { - dispatch_to_app(request, api_tx, timeout, None, None, None) + dispatch_to_app(request, api_tx, timeout, None, None) } pub(super) fn dispatch_to_app_with_caller_timeout( @@ -941,55 +938,24 @@ pub(super) fn dispatch_to_app_with_caller_timeout( api_tx, timeout, None, - None, Some(("timeout", "timed out waiting for agent status")), ) } -pub(super) fn dispatch_stream_open( - request: Request, - api_tx: &ApiRequestSender, - timeout: Duration, - active: Arc<AtomicBool>, -) -> String { - dispatch_to_app(request, api_tx, Some(timeout), None, Some(active), None) -} - -pub(super) fn dispatch_stream_frame( - request: Request, - api_tx: &ApiRequestSender, - active: Arc<AtomicBool>, -) -> String { - dispatch_to_app( - request, - api_tx, - Some(crate::app::pane_graphics::DIRECT_OUTER_TIMEOUT), - None, - Some(active), - None, - ) -} - fn dispatch_to_app( request: Request, api_tx: &ApiRequestSender, timeout: Option<Duration>, response_write_complete: Option<std::sync::mpsc::Receiver<()>>, - stream_active: Option<Arc<AtomicBool>>, timeout_response: Option<(&str, &str)>, ) -> String { let request_id = request.id.clone(); - let request_active = stream_active.clone(); let (respond_to, response_rx) = std::sync::mpsc::channel(); if let Err(err) = api_tx.send(ApiRequestMessage { request, respond_to, response_write_complete, - stream_active, }) { - if let Some(active) = request_active { - active.store(false, Ordering::Release); - } return error_response_json( request_id, "server_unavailable", @@ -1019,9 +985,6 @@ fn dispatch_to_app( match response { Ok(response) => response, Err(err) => { - if let Some(active) = request_active { - active.store(false, Ordering::Release); - } if err.kind() == std::io::ErrorKind::TimedOut { if let Some((code, message)) = timeout_response { return error_response_json(request_id, code, message.into()); @@ -1300,6 +1263,93 @@ mod tests { } #[test] + fn removed_pane_graphics_methods_return_unknown_method_without_stream_upgrade() { + for method in [ + "pane.graphics.info", + "pane.graphics.set", + "pane.graphics.clear", + "pane.graphics.stream", + ] { + let (mut client, server, _path) = local_stream_pair("removed-pane-graphics"); + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + writeln!( + client, + "{{\"id\":\"removed\",\"method\":\"{method}\",\"params\":{{}}}}" + ) + .unwrap(); + client.flush().unwrap(); + + handle_connection( + server, + &api_tx, + &EventHub::default(), + &Arc::new(AtomicBool::new(true)), + None, + ) + .unwrap(); + + let response = read_line(&mut client); + let response: serde_json::Value = serde_json::from_str(&response).unwrap(); + assert_eq!(response["id"], "removed", "{method}"); + assert_eq!(response["error"]["code"], "unknown_method", "{method}"); + assert!(response["error"]["message"] + .as_str() + .is_some_and(|message| message.contains(method))); + assert!(response.get("result").is_none(), "{method}"); + assert!(api_rx.try_recv().is_err(), "{method} reached the app"); + } + } + + #[test] + fn unrelated_unknown_method_retains_standard_invalid_request_response() { + let (mut client, server, _path) = local_stream_pair("unknown-api-request"); + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + client + .write_all(b"{\"id\":\"unknown\",\"method\":\"nope\",\"params\":{}}\n") + .unwrap(); + client.flush().unwrap(); + + handle_connection( + server, + &api_tx, + &EventHub::default(), + &Arc::new(AtomicBool::new(true)), + None, + ) + .unwrap(); + + let response = read_line(&mut client); + let response: serde_json::Value = serde_json::from_str(&response).unwrap(); + assert_eq!(response["id"], "unknown"); + assert_eq!(response["error"]["code"], "invalid_request"); + assert!(api_rx.try_recv().is_err()); + } + + #[test] + fn ordinary_api_request_still_uses_normal_connection_path() { + let (mut client, server, _path) = local_stream_pair("ordinary-api-request"); + let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + client + .write_all(b"{\"id\":\"ordinary\",\"method\":\"ping\",\"params\":{}}\n") + .unwrap(); + client.flush().unwrap(); + + handle_connection( + server, + &api_tx, + &EventHub::default(), + &Arc::new(AtomicBool::new(true)), + None, + ) + .unwrap(); + + let response = read_line(&mut client); + let response: serde_json::Value = serde_json::from_str(&response).unwrap(); + assert_eq!(response["id"], "ordinary"); + assert_eq!(response["result"]["type"], "pong"); + } + + #[test] fn ping_request_returns_pong() { let (tx, _rx) = mpsc::unbounded_channel(); let response = handle_request( @@ -1758,39 +1808,3 @@ 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(), - layer_id: None, - z_index: 0, - 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 deleted file mode 100644 index 55816b1d..00000000 --- a/src/api/server/pane_graphics_stream.rs +++ /dev/null @@ -1,1114 +0,0 @@ -use std::io::{self, Read}; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::Arc; -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_stream_frame, dispatch_stream_open, - 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); - -#[derive(serde::Deserialize)] -struct FrameFile { - path: String, -} - -#[derive(serde::Deserialize)] -struct FrameHeader { - format: crate::api::schema::PaneGraphicsFormat, - image_width: u32, - image_height: u32, - #[serde(default)] - data_length: Option<usize>, - #[serde(default)] - file: Option<FrameFile>, - #[serde(default)] - sequence: u64, - #[serde(default)] - revision: u64, - #[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 layer_id = params.layer_id.clone(); - let z_index = params.z_index; - let owner = next_owner(); - params.owner = owner.clone(); - let stream_active = Arc::new(AtomicBool::new(true)); - let open_response = dispatch_stream_open( - Request { - id: request_id.clone(), - method: Method::PaneGraphicsStreamOpen(params), - }, - api_tx, - open_timeout, - Arc::clone(&stream_active), - ); - if api_response_outcome(&open_response) != "ok" { - stream_active.store(false, Ordering::Release); - let write_result = write_text_line_allow_disconnect(&mut stream, &open_response); - clear_layer(&pane_id, layer_id.as_deref(), z_index, &owner, api_tx); - write_result?; - return Ok(()); - } - - if let Err(err) = write_json_line( - &mut stream, - &SuccessResponse { - id: request_id.clone(), - result: ResponseResult::Ok {}, - }, - ) { - stream_active.store(false, Ordering::Release); - clear_layer(&pane_id, layer_id.as_deref(), z_index, &owner, api_tx); - if is_connection_closed_error(&err) { - return Ok(()); - } - return Err(err); - } - - let result = serve_frames( - &mut stream, - &request_id, - &owner, - &pane_id, - layer_id.as_deref(), - z_index, - api_tx, - running, - &stream_active, - read_timeouts, - ); - stream_active.store(false, Ordering::Release); - clear_layer(&pane_id, layer_id.as_deref(), z_index, &owner, api_tx); - result -} - -fn serve_frames( - stream: &mut LocalStream, - request_id: &str, - owner: &str, - pane_id: &str, - layer_id: Option<&str>, - z_index: i32, - 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 let Some(file) = header.file { - if !matches!( - header.format, - crate::api::schema::PaneGraphicsFormat::Rgba - | crate::api::schema::PaneGraphicsFormat::Bgra - ) { - write_json_line_allow_disconnect( - stream, - &ErrorResponse { - id: request_id.to_string(), - error: ErrorBody { - code: "invalid_frame".into(), - message: "file frames require rgba or bgra".into(), - }, - }, - )?; - return Ok(()); - } - let response = dispatch_stream_frame( - Request { - id: format!("{request_id}:file:{}", header.sequence), - method: Method::PaneGraphicsStreamDirect( - crate::api::schema::PaneGraphicsDirectParams { - pane_id: pane_id.to_owned(), - layer_id: layer_id.map(str::to_owned), - z_index, - owner: owner.to_owned(), - image_width: header.image_width, - image_height: header.image_height, - format: header.format, - path: file.path, - sequence: header.sequence, - revision: header.revision, - placement: header.placement, - }, - ), - }, - api_tx, - Arc::clone(stream_active), - ); - write_text_line_allow_disconnect(stream, &response)?; - if api_response_outcome(&response) != "ok" { - return Ok(()); - } - continue; - } - - let Some(data_length) = header.data_length else { - write_json_line_allow_disconnect( - stream, - &ErrorResponse { - id: request_id.to_string(), - error: ErrorBody { - code: "invalid_frame".into(), - message: "frame requires data_length or file".into(), - }, - }, - )?; - return Ok(()); - }; - if 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 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, - 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(), - layer_id: layer_id.map(str::to_owned), - z_index, - 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_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, - layer_id: Option<&str>, - z_index: i32, - 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(), - layer_id: layer_id.map(str::to_owned), - z_index, - 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 wait.read(stream, &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 wait.read(stream, &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 read(&self, stream: &mut LocalStream, buffer: &mut [u8]) -> io::Result<usize> { - if matches!(self, Self::SocketTimeout) { - return stream.read(buffer); - } - match crate::ipc::poll_local_stream_read_count(stream, buffer)? { - crate::ipc::LocalStreamReadCount::Data(count) => Ok(count), - crate::ipc::LocalStreamReadCount::Pending => Err(io::ErrorKind::WouldBlock.into()), - crate::ipc::LocalStreamReadCount::Closed => Ok(0), - } - } - - 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<Option<T>>, -) -> std::io::Result<Option<T>> { - match stream.set_recv_timeout(Some(CONNECTION_POLL_INTERVAL)) { - Ok(()) => { - let result = read(stream, ReadWait::SocketTimeout); - finish_timed_read(result, || stream.set_recv_timeout(None)) - } - Err(err) if err.kind() == io::ErrorKind::Unsupported => { - crate::ipc::set_local_stream_polling(stream, true)?; - let result = read(stream, ReadWait::Poll(PollBackoff::new())); - finish_timed_read(result, || { - crate::ipc::set_local_stream_polling(stream, false) - }) - } - // A peer can disconnect after the caller's running check but before - // setsockopt. macOS reports that closed-socket race as EINVAL. - Err(err) if err.kind() == io::ErrorKind::InvalidInput => Ok(None), - Err(err) => Err(err), - } -} - -fn finish_timed_read<T>( - result: std::io::Result<Option<T>>, - reset: impl FnOnce() -> std::io::Result<()>, -) -> std::io::Result<Option<T>> { - match result { - // None is terminal for this dedicated stream. macOS returns EINVAL when - // socket options are restored after the peer has already disconnected. - Ok(None) => Ok(None), - Ok(value) => { - reset()?; - Ok(value) - } - Err(err) => { - let _ = reset(); - 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; - #[cfg(unix)] - use crate::api::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, AtomicU64, Ordering}; - use std::sync::Arc; - use std::time::{Duration, Instant}; - use tokio::sync::mpsc; - - static NEXT_LOCAL_STREAM_ID: AtomicU64 = AtomicU64::new(1); - - fn local_stream_pair(_name: &str) -> (LocalStream, LocalStream, PathBuf) { - let unique = format!( - "hpg-{}-{}.sock", - std::process::id(), - NEXT_LOCAL_STREAM_ID.fetch_add(1, Ordering::Relaxed) - ); - 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 - } - - #[cfg(unix)] - fn assert_server_stream_owner(owner: &str) { - assert!(owner.starts_with("pane.graphics.stream:")); - } - - fn respond_ok(message: ApiRequestMessage) { - let response = serde_json::to_string(&SuccessResponse { - id: message.request.id, - result: ResponseResult::Ok {}, - }) - .unwrap(); - message.respond_to.send(response).unwrap(); - } - - fn assert_close_and_respond(message: ApiRequestMessage, pane_id: &str, owner: &str) { - match &message.request.method { - Method::PaneGraphicsStreamClose(params) => { - assert_eq!(params.pane_id, pane_id); - assert_eq!(params.owner, owner); - } - other => panic!("unexpected close request: {other:?}"), - } - respond_ok(message); - } - - #[test] - fn browser_file_header_accepts_damage_but_keeps_full_canonical_frame() { - let header: FrameHeader = serde_json::from_str( - r#"{"format":"rgba","image_width":2,"image_height":3,"sequence":7,"revision":8,"file":{"path":"/private/frame"},"damage":{"x":1,"y":1,"width":1,"height":1},"transport":"direct-kitty","placement":{"grid_cols":2,"grid_rows":3}}"#, - ) - .unwrap(); - assert_eq!(header.data_length, None); - assert_eq!(header.file.unwrap().path, "/private/frame"); - assert_eq!((header.sequence, header.revision), (7, 8)); - assert_eq!((header.image_width, header.image_height), (2, 3)); - assert_eq!( - (header.placement.grid_cols, header.placement.grid_rows), - (2, 3) - ); - } - - #[test] - fn timed_read_skips_reset_after_stream_ends() { - let mut reset_called = false; - let result = finish_timed_read::<()>(Ok(None), || { - reset_called = true; - Ok(()) - }); - - assert!(result.unwrap().is_none()); - assert!(!reset_called); - } - - #[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:?}"), - }; - respond_ok(open); - - 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:?}"), - } - respond_ok(msg); - - drop(client); - running.store(false, Ordering::Relaxed); - assert_close_and_respond(api_rx.blocking_recv().unwrap(), "pane_1", &stream_owner); - 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 are disabled by terminal.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"); - - assert_close_and_respond(api_rx.blocking_recv().unwrap(), "pane_1", &stream_owner); - - 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(), - layer_id: None, - z_index: 0, - 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")); - - assert_close_and_respond(api_rx.blocking_recv().unwrap(), "pane_1", &stream_owner); - - 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); - respond_ok(open); - - 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(); - assert_close_and_respond(close, "pane_1", &stream_owner); - - 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 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)); - } - - #[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(), - layer_id: None, - z_index: 0, - 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:?}"), - }; - respond_ok(open); - 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(); - assert_close_and_respond(close, "pane_1", &owner); - - 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", - None, - 0, - &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()); - } -} |