diff options
Diffstat (limited to 'src/api/server/pane_graphics_stream.rs')
| -rw-r--r-- | src/api/server/pane_graphics_stream.rs | 1114 |
1 files changed, 0 insertions, 1114 deletions
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()); - } -} |