From cc88b3b8e5bb9f7d9f23ed6ae85a52fd7b5b9ed6 Mon Sep 17 00:00:00 2001 From: Can Celik Date: Wed, 2 Sep 2026 02:45:20 +0300 Subject: feat: add stable client endpoint compatibility (#3509) * feat: add stable client endpoint compatibility * fix: keep portable-pty patch checks composable * ci: run maintenance checks in pull requests * fix: reject endpoint errors from terminal clients * feat: keep compatible servers running across updates --- src/server/autodetect.rs | 15 +- src/server/client_commands.rs | 240 +++++++++++++++--- src/server/client_transport.rs | 477 +++++++++++++++++++++++++++-------- src/server/headless.rs | 35 ++- src/server/headless/render.rs | 11 +- src/server/headless/tests/mod.rs | 109 +++++--- src/server/render_scale_benchmark.rs | 44 ++++ 7 files changed, 733 insertions(+), 198 deletions(-) (limited to 'src/server') diff --git a/src/server/autodetect.rs b/src/server/autodetect.rs index 1813ebd9..bd364a70 100644 --- a/src/server/autodetect.rs +++ b/src/server/autodetect.rs @@ -150,19 +150,22 @@ fn validate_running_server_compatibility() -> io::Result<()> { ))); }; - if status.protocol == Some(crate::protocol::PROTOCOL_VERSION) { + let endpoint_generation = status + .capabilities + .as_ref() + .and_then(|capabilities| capabilities.endpoint_protocol_generation); + if endpoint_generation == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION) { return Ok(()); } Err(io::Error::other(format!( - "Herdr was updated, but this session is still running the old server.\n\nserver: v{} protocol {}\nclient: v{} protocol {}\n\n{}", + "This session predates Herdr's stable endpoint protocol and needs one final server update.\n\nserver: v{} endpoint generation {}\nclient: v{} endpoint generation {}\n\n{}", status.version.as_deref().unwrap_or("unknown"), - status - .protocol + endpoint_generation .map(|value| value.to_string()) - .unwrap_or_else(|| "unknown".to_string()), + .unwrap_or_else(|| "unavailable".to_string()), crate::build_info::version(), - crate::protocol::PROTOCOL_VERSION, + crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION, crate::session::active_restart_after_update_guidance() ))) } diff --git a/src/server/client_commands.rs b/src/server/client_commands.rs index eeb49c7e..86243e31 100644 --- a/src/server/client_commands.rs +++ b/src/server/client_commands.rs @@ -12,47 +12,56 @@ pub(crate) const MAX_ENDPOINT_BOOT_ID_BYTES: usize = 128; pub(crate) const MAX_ENDPOINT_REQUEST_ID_BYTES: usize = 128; const ENDPOINT_RESPONSE_CHUNK_BYTES: usize = 512 * 1024; +const CLIENT_SHELL_METHODS: &[&str] = &[ + "command.invoke", + "integration.install", + "integration.list", + "layout.set_split_ratio", + "pane.close", + "pane.copy_motion", + "pane.copy_search", + "pane.edit_scrollback", + "pane.focus", + "pane.focus_direction", + "pane.input.set", + "pane.link.activate", + "pane.rename", + "pane.resize", + "pane.scroll", + "pane.selection.read", + "pane.split", + "pane.swap", + "pane.zoom", + "product_announcement.dismiss", + "release_notes.dismiss", + "server.reload_config", + "tab.close", + "tab.create", + "tab.focus", + "tab.move", + "tab.rename", + "workspace.close", + "workspace.create", + "workspace.focus", + "workspace.move", + "workspace.move_block", + "workspace.rename", + "worktree.create", + "worktree.list", + "worktree.open", + "worktree.remove", +]; + +pub(crate) fn supported_client_shell_method_names() -> &'static [&'static str] { + CLIENT_SHELL_METHODS +} + +pub(crate) fn supports_client_shell_method_name(method: &str) -> bool { + CLIENT_SHELL_METHODS.contains(&method) +} + pub(crate) fn supports_client_shell_method(method: &Method) -> bool { - matches!( - method, - Method::CommandInvoke(_) - | Method::IntegrationInstall(_) - | Method::IntegrationList(_) - | Method::LayoutSetSplitRatio(_) - | Method::PaneClose(_) - | Method::PaneCopyMotion(_) - | Method::PaneCopySearch(_) - | Method::PaneEditScrollback(_) - | Method::PaneFocus(_) - | Method::PaneFocusDirection(_) - | Method::PaneInputSet(_) - | Method::PaneLinkActivate(_) - | Method::PaneRename(_) - | Method::PaneResize(_) - | Method::PaneScroll(_) - | Method::PaneSelectionRead(_) - | Method::PaneSplit(_) - | Method::PaneSwap(_) - | Method::PaneZoom(_) - | Method::ProductAnnouncementDismiss(_) - | Method::ReleaseNotesDismiss(_) - | Method::ServerReloadConfig(_) - | Method::TabClose(_) - | Method::TabCreate(_) - | Method::TabFocus(_) - | Method::TabMove(_) - | Method::TabRename(_) - | Method::WorkspaceClose(_) - | Method::WorkspaceCreate(_) - | Method::WorkspaceFocus(_) - | Method::WorkspaceMove(_) - | Method::WorkspaceMoveBlock(_) - | Method::WorkspaceRename(_) - | Method::WorktreeCreate(_) - | Method::WorktreeList(_) - | Method::WorktreeOpen(_) - | Method::WorktreeRemove(_) - ) + supports_client_shell_method_name(crate::api::api_method_name(method)) } pub(crate) fn error_response(id: String, code: &str, message: impl Into) -> String { @@ -134,8 +143,159 @@ pub(crate) fn spawn_response_waiter( #[cfg(test)] mod tests { + use std::collections::{BTreeMap, BTreeSet}; + + use sha2::{Digest, Sha256}; + use super::*; + fn collect_schema_refs(value: &serde_json::Value, refs: &mut BTreeSet) { + match value { + serde_json::Value::Object(object) => { + if let Some(reference) = object.get("$ref").and_then(serde_json::Value::as_str) { + if let Some(name) = reference.rsplit('/').next() { + refs.insert(name.to_owned()); + } + } + for value in object.values() { + collect_schema_refs(value, refs); + } + } + serde_json::Value::Array(values) => { + for value in values { + collect_schema_refs(value, refs); + } + } + _ => {} + } + } + + fn normalized_wire_schema(value: &serde_json::Value) -> serde_json::Value { + match value { + serde_json::Value::Object(object) => serde_json::Value::Object( + object + .iter() + .filter(|(key, _)| { + !matches!( + key.as_str(), + "description" | "examples" | "readOnly" | "title" | "writeOnly" + ) + }) + .map(|(key, value)| (key.clone(), normalized_wire_schema(value))) + .collect(), + ), + serde_json::Value::Array(values) => { + serde_json::Value::Array(values.iter().map(normalized_wire_schema).collect()) + } + _ => value.clone(), + } + } + + fn endpoint_method_shape_digests() -> BTreeMap { + let schema = serde_json::to_value(schemars::schema_for!(crate::api::schema::Request)) + .expect("request schema"); + let definitions = schema + .get("$defs") + .and_then(serde_json::Value::as_object) + .expect("request definitions"); + let branches = schema + .get("oneOf") + .and_then(serde_json::Value::as_array) + .expect("request method branches"); + let mut digests = BTreeMap::new(); + + for method in CLIENT_SHELL_METHODS { + let branch = branches + .iter() + .find(|branch| { + branch + .pointer("/properties/method/const") + .and_then(serde_json::Value::as_str) + == Some(method) + }) + .unwrap_or_else(|| panic!("missing request schema branch for {method}")); + let mut referenced_names = BTreeSet::new(); + collect_schema_refs(branch, &mut referenced_names); + let mut visited_names = BTreeSet::new(); + let mut selected_definitions = serde_json::Map::new(); + while let Some(name) = referenced_names.pop_first() { + if !visited_names.insert(name.clone()) { + continue; + } + let definition = definitions + .get(&name) + .unwrap_or_else(|| panic!("missing schema definition {name} for {method}")); + collect_schema_refs(definition, &mut referenced_names); + selected_definitions.insert(name, normalized_wire_schema(definition)); + } + let shape = serde_json::json!({ + "request": normalized_wire_schema(branch), + "definitions": selected_definitions, + }); + let bytes = serde_json::to_vec(&shape).expect("method shape json"); + digests.insert(method.to_string(), format!("{:x}", Sha256::digest(bytes))); + } + + digests + } + + #[test] + fn advertised_client_shell_method_shapes_stay_at_the_v1_contract() { + let expected: BTreeMap = serde_json::from_str(include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/endpoint-method-shapes-v1.json" + ))) + .expect("endpoint method shape fixture"); + let actual = endpoint_method_shape_digests(); + + assert_eq!( + actual, + expected, + "an existing endpoint method changed shape; add load-bearing behavior as a new advertised method or explicitly gate new fields" + ); + } + + #[test] + fn advertised_client_shell_methods_are_sorted_unique_and_in_schema() { + assert!(CLIENT_SHELL_METHODS + .windows(2) + .all(|pair| pair[0] < pair[1])); + + fn collect_method_constants(value: &serde_json::Value, methods: &mut Vec) { + match value { + serde_json::Value::Object(object) => { + if let Some(method) = object + .get("const") + .and_then(serde_json::Value::as_str) + .filter(|value| value.contains('.')) + { + methods.push(method.to_owned()); + } + for value in object.values() { + collect_method_constants(value, methods); + } + } + serde_json::Value::Array(values) => { + for value in values { + collect_method_constants(value, methods); + } + } + _ => {} + } + } + + let schema = serde_json::to_value(schemars::schema_for!(crate::api::schema::Request)) + .expect("request schema"); + let mut schema_methods = Vec::new(); + collect_method_constants(&schema, &mut schema_methods); + for method in CLIENT_SHELL_METHODS { + assert!( + schema_methods.iter().any(|candidate| candidate == method), + "advertised endpoint method {method:?} is absent from the request schema" + ); + } + } + #[test] fn client_shell_lane_excludes_api_front_door_and_lifecycle_methods() { assert!(supports_client_shell_method(&Method::ServerReloadConfig( diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs index 673e16d3..25a963e7 100644 --- a/src/server/client_transport.rs +++ b/src/server/client_transport.rs @@ -17,6 +17,10 @@ use tokio::sync::mpsc; use tracing::{debug, warn}; use crate::ipc::LocalStream; +use crate::protocol::endpoint::{ + EndpointClientHello, EndpointServerWelcome, ENDPOINT_HELLO_KIND, ENDPOINT_PROTOCOL_GENERATION, + ENDPOINT_WELCOME_KIND, +}; use crate::protocol::{ self, AttachScrollDirection, AttachScrollSource, ClientMessage, ClientPaneInputEvent, RenderEncoding, ServerMessage, MAX_CLIPBOARD_IMAGE_PAYLOAD, MAX_FRAME_SIZE, @@ -39,6 +43,74 @@ const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(4); /// Maximum input payload size (bytes) for a single `ClientMessage::Input`. const MAX_INPUT_PAYLOAD: usize = 1024 * 1024; // 1 MB +const MAX_CLIENT_SHELL_DIMENSION: u16 = 4096; +const MAX_CLIENT_SHELL_CELLS: u32 = 1_000_000; +const MAX_CLIENT_CELL_SIZE_PX: u32 = 4096; + +fn client_shell_geometry_error( + surface_size: crate::protocol::ClientSurfaceSize, + cell_width_px: u32, + cell_height_px: u32, +) -> Option<&'static str> { + if surface_size.cols == 0 || surface_size.rows == 0 { + return Some("client shell requires a non-empty pane surface"); + } + if surface_size.cols > MAX_CLIENT_SHELL_DIMENSION + || surface_size.rows > MAX_CLIENT_SHELL_DIMENSION + || u32::from(surface_size.cols) * u32::from(surface_size.rows) > MAX_CLIENT_SHELL_CELLS + { + return Some("client shell pane surface exceeds the safe geometry limit"); + } + if cell_width_px > MAX_CLIENT_CELL_SIZE_PX || cell_height_px > MAX_CLIENT_CELL_SIZE_PX { + return Some("client shell cell pixel size exceeds the safe geometry limit"); + } + None +} + +#[derive(serde::Deserialize)] +struct EndpointRequestHead { + id: String, + method: String, +} + +enum DecodedEndpointRequest { + Dispatch(Box), + Error { + request_id: String, + code: &'static str, + message: String, + }, +} + +fn write_endpoint_rejection(stream: &mut LocalStream, code: &str, message: impl Into) { + let welcome = EndpointServerWelcome::incompatible(code, message); + let response = ServerMessage::EndpointControl { + kind: ENDPOINT_WELCOME_KIND.into(), + data: serde_json::to_string(&welcome).unwrap_or_else(|_| "{}".into()), + }; + let _ = protocol::write_message(stream, &response); +} + +fn decode_endpoint_request(request: &str) -> serde_json::Result { + let head = serde_json::from_str::(request)?; + if !crate::server::client_commands::supports_client_shell_method_name(&head.method) { + return Ok(DecodedEndpointRequest::Error { + request_id: head.id, + code: "unsupported_method", + message: format!("method {:?} is not available on this machine", head.method), + }); + } + Ok( + match serde_json::from_str::(request) { + Ok(request) => DecodedEndpointRequest::Dispatch(Box::new(request)), + Err(error) => DecodedEndpointRequest::Error { + request_id: head.id, + code: "invalid_request", + message: format!("invalid endpoint request: {error}"), + }, + }, + ) +} /// Maximum structured input events accepted in one client message. const MAX_INPUT_EVENT_BATCH: usize = 4096; @@ -417,6 +489,14 @@ pub(crate) enum ServerEvent { boot_id: String, request: Box, }, + /// A well-framed endpoint request could not be dispatched by this server. + ClientShellEndpointRequestError { + client_id: u64, + boot_id: String, + request_id: String, + code: &'static str, + message: String, + }, /// One chunk of a deferred endpoint operation's final response is ready. ClientShellEndpointResponseChunkReady { client_id: u64, @@ -616,50 +696,69 @@ pub(crate) fn handle_client_handshake( let (cols, rows) = clamp_terminal_size(cols, rows); (cols, rows, cell_width_px, cell_height_px, pixel_mouse, None) } - ClientMessage::ClientShellHello { - version, - cell_width_px, - cell_height_px, - surface_size, - pixel_mouse, - direct_graphics, - endpoint_keybindings, - mouse_capture, - } => { - if let protocol::VersionCheck::Incompatible(reason) = - protocol::check_client_version(version) - { - let welcome = ServerMessage::Welcome { - version: PROTOCOL_VERSION, - encoding: RenderEncoding::SemanticFrame, - error: Some(reason), - }; - let _ = protocol::write_message(&mut stream, &welcome); - return Ok(()); - } - if surface_size.cols == 0 || surface_size.rows == 0 { - let welcome = ServerMessage::Welcome { - version: PROTOCOL_VERSION, - encoding: RenderEncoding::SemanticFrame, - error: Some("client shell requires a non-empty pane surface".to_owned()), - }; - let _ = protocol::write_message(&mut stream, &welcome); + ClientMessage::EndpointControl { kind, data } if kind == ENDPOINT_HELLO_KIND => { + let hello: EndpointClientHello = match serde_json::from_str(&data) { + Ok(hello) => hello, + Err(error) => { + write_endpoint_rejection( + &mut stream, + "invalid_hello", + format!("invalid endpoint hello: {error}"), + ); + return Ok(()); + } + }; + let incompatibility = if hello.generation != ENDPOINT_PROTOCOL_GENERATION { + Some(( + "unsupported_generation", + format!( + "endpoint generation {} is unsupported; this server supports generation {ENDPOINT_PROTOCOL_GENERATION}", + hello.generation + ), + )) + } else if !hello.supports_required_codecs() { + Some(( + "no_common_core", + "client and server have no compatible endpoint core codecs".to_owned(), + )) + } else { + client_shell_geometry_error( + hello.surface_size, + hello.cell_width_px, + hello.cell_height_px, + ) + .map(|reason| ("invalid_surface", reason.to_owned())) + }; + if let Some((code, reason)) = incompatibility { + write_endpoint_rejection(&mut stream, code, reason); return Ok(()); } ( - surface_size.cols, - surface_size.rows, - cell_width_px, - cell_height_px, + hello.surface_size.cols, + hello.surface_size.rows, + hello.cell_width_px, + hello.cell_height_px, false, Some(( - pixel_mouse, - direct_graphics, - endpoint_keybindings, - mouse_capture, + hello.pixel_mouse, + hello.direct_graphics, + hello.endpoint_keybindings, + hello.mouse_capture, )), ) } + ClientMessage::ClientShellHello { .. } => { + let welcome = ServerMessage::Welcome { + version: PROTOCOL_VERSION, + encoding: RenderEncoding::SemanticFrame, + error: Some( + "this client predates the stable endpoint protocol; upgrade the Herdr client" + .to_owned(), + ), + }; + let _ = protocol::write_message(&mut stream, &welcome); + return Ok(()); + } _ => { debug!(client_id, "first message was not a handshake, closing"); let welcome = ServerMessage::Welcome { @@ -678,16 +777,30 @@ pub(crate) fn handle_client_handshake( return Ok(()); } - // Send Welcome. + // Send the negotiated welcome. Endpoint compatibility is independent from + // the same-install protocol used by direct terminal clients. let render_encoding = if shell_options.is_some() { RenderEncoding::SemanticFrame } else { RenderEncoding::TerminalAnsi }; - let welcome = ServerMessage::Welcome { - version: PROTOCOL_VERSION, - encoding: render_encoding, - error: None, + let welcome = if shell_options.is_some() { + let welcome = EndpointServerWelcome::compatible( + crate::server::client_commands::supported_client_shell_method_names() + .iter() + .map(|method| (*method).to_owned()) + .collect(), + ); + ServerMessage::EndpointControl { + kind: ENDPOINT_WELCOME_KIND.into(), + data: serde_json::to_string(&welcome).map_err(io::Error::other)?, + } + } else { + ServerMessage::Welcome { + version: PROTOCOL_VERSION, + encoding: render_encoding, + error: None, + } }; protocol::write_message(&mut stream, &welcome).map_err(|e| io::Error::other(e.to_string()))?; @@ -947,14 +1060,24 @@ fn client_read_loop( cell_height_px, surface_size, pixel_mouse, - } => ServerEvent::ClientShellResize { - client_id, - surface_cols: surface_size.cols.max(1), - surface_rows: surface_size.rows.max(1), - cell_width_px, - cell_height_px, - pixel_mouse, - }, + } => { + if let Some(reason) = + client_shell_geometry_error(surface_size, cell_width_px, cell_height_px) + { + warn!(client_id, %reason, "invalid client shell resize, closing"); + let _ = server_event_tx + .blocking_send(ServerEvent::ClientDisconnected { client_id }); + break; + } + ServerEvent::ClientShellResize { + client_id, + surface_cols: surface_size.cols, + surface_rows: surface_size.rows, + cell_width_px, + cell_height_px, + pixel_mouse, + } + } ClientMessage::ClientShellHostTheme { update } => { if matches!( &update, @@ -1075,14 +1198,20 @@ fn client_read_loop( .blocking_send(ServerEvent::ClientDisconnected { client_id }); break; } - let Ok(request) = serde_json::from_str::(&request) - else { - warn!(client_id, "invalid client shell endpoint command, closing"); - let _ = server_event_tx - .blocking_send(ServerEvent::ClientDisconnected { client_id }); - break; + let decoded = match decode_endpoint_request(&request) { + Ok(decoded) => decoded, + Err(error) => { + warn!(client_id, %error, "invalid endpoint request envelope, closing"); + let _ = server_event_tx + .blocking_send(ServerEvent::ClientDisconnected { client_id }); + break; + } }; - if request.id.len() > crate::server::client_commands::MAX_ENDPOINT_REQUEST_ID_BYTES + let request_id = match &decoded { + DecodedEndpointRequest::Dispatch(request) => request.id.as_str(), + DecodedEndpointRequest::Error { request_id, .. } => request_id, + }; + if request_id.len() > crate::server::client_commands::MAX_ENDPOINT_REQUEST_ID_BYTES { warn!( client_id, @@ -1092,12 +1221,31 @@ fn client_read_loop( .blocking_send(ServerEvent::ClientDisconnected { client_id }); break; } - ServerEvent::ClientShellEndpointRequest { - client_id, - boot_id, - request: Box::new(request), + match decoded { + DecodedEndpointRequest::Dispatch(request) => { + ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id, + request, + } + } + DecodedEndpointRequest::Error { + request_id, + code, + message, + } => ServerEvent::ClientShellEndpointRequestError { + client_id, + boot_id, + request_id, + code, + message, + }, } } + ClientMessage::EndpointControl { kind, .. } => { + debug!(client_id, %kind, "ignoring unknown endpoint control message"); + continue; + } ClientMessage::Detach => { let _ = server_event_tx.blocking_send(ServerEvent::ClientDetach { client_id }); break; @@ -1195,6 +1343,38 @@ mod tests { (client, server, TestSocketPath(path)) } + fn endpoint_hello(surface_cols: u16, surface_rows: u16) -> ClientMessage { + let hello = EndpointClientHello { + generation: ENDPOINT_PROTOCOL_GENERATION, + cell_width_px: 8, + cell_height_px: 16, + surface_size: crate::protocol::ClientSurfaceSize { + cols: surface_cols, + rows: surface_rows, + }, + pixel_mouse: true, + direct_graphics: true, + endpoint_keybindings: true, + mouse_capture: true, + snapshot_codecs: vec![crate::protocol::endpoint::SNAPSHOT_CODEC_V1.into()], + surface_codecs: vec![crate::protocol::endpoint::SURFACE_CODEC_V1.into()], + input_codecs: vec![crate::protocol::endpoint::INPUT_CODEC_V1.into()], + blob_codecs: vec![crate::protocol::endpoint::BLOB_CODEC_V1.into()], + }; + ClientMessage::EndpointControl { + kind: ENDPOINT_HELLO_KIND.into(), + data: serde_json::to_string(&hello).unwrap(), + } + } + + fn endpoint_welcome(message: ServerMessage) -> EndpointServerWelcome { + let ServerMessage::EndpointControl { kind, data } = message else { + panic!("expected endpoint welcome"); + }; + assert_eq!(kind, ENDPOINT_WELCOME_KIND); + serde_json::from_str(&data).unwrap() + } + fn recv_server_event(receiver: &mut mpsc::Receiver, context: &str) -> ServerEvent { let deadline = std::time::Instant::now() + Duration::from_secs(1); loop { @@ -1431,6 +1611,62 @@ mod tests { ); } + #[test] + fn client_shell_geometry_rejects_unsafe_dimensions_and_cell_sizes() { + assert!(client_shell_geometry_error( + crate::protocol::ClientSurfaceSize { cols: 80, rows: 24 }, + 8, + 16, + ) + .is_none()); + assert!(client_shell_geometry_error( + crate::protocol::ClientSurfaceSize { + cols: MAX_CLIENT_SHELL_DIMENSION, + rows: MAX_CLIENT_SHELL_DIMENSION, + }, + 8, + 16, + ) + .is_some()); + assert!(client_shell_geometry_error( + crate::protocol::ClientSurfaceSize { cols: 80, rows: 24 }, + MAX_CLIENT_CELL_SIZE_PX + 1, + 16, + ) + .is_some()); + } + + #[test] + fn unknown_endpoint_method_returns_correlated_error() { + let decoded = decode_endpoint_request( + r#"{"id":"req-1","method":"plugin.future","params":{"value":1}}"#, + ) + .unwrap(); + assert!(matches!( + decoded, + DecodedEndpointRequest::Error { + request_id, + code: "unsupported_method", + .. + } if request_id == "req-1" + )); + } + + #[test] + fn malformed_known_endpoint_method_returns_correlated_error() { + let decoded = + decode_endpoint_request(r#"{"id":"req-2","method":"workspace.focus","params":{}}"#) + .unwrap(); + assert!(matches!( + decoded, + DecodedEndpointRequest::Error { + request_id, + code: "invalid_request", + .. + } if request_id == "req-2" + )); + } + #[test] fn handshake_negotiates_terminal_ansi_encoding() { let (mut client_stream, server_stream, _path) = local_stream_pair("client-handshake-ansi"); @@ -1509,31 +1745,14 @@ mod tests { handle_client_handshake(server_stream, 43, &server_event_tx, &handshake_quit) }); - protocol::write_message( - &mut client_stream, - &ClientMessage::ClientShellHello { - version: PROTOCOL_VERSION, - cell_width_px: 8, - cell_height_px: 16, - surface_size: crate::protocol::ClientSurfaceSize { cols: 80, rows: 29 }, - pixel_mouse: true, - direct_graphics: true, - endpoint_keybindings: true, - mouse_capture: true, - }, - ) - .expect("write shell hello"); + protocol::write_message(&mut client_stream, &endpoint_hello(80, 29)) + .expect("write shell hello"); let welcome: ServerMessage = protocol::read_message(&mut client_stream, MAX_FRAME_SIZE).expect("read welcome"); - assert!(matches!( - welcome, - ServerMessage::Welcome { - encoding: RenderEncoding::SemanticFrame, - error: None, - .. - } - )); + let welcome = endpoint_welcome(welcome); + assert_eq!(welcome.generation, ENDPOINT_PROTOCOL_GENERATION); + assert!(welcome.error.is_none()); match server_event_rx .blocking_recv() .expect("client shell connected event") @@ -1581,31 +1800,15 @@ mod tests { handle_client_handshake(server_stream, 43, &server_event_tx, &handshake_quit) }); - protocol::write_message( - &mut client_stream, - &ClientMessage::ClientShellHello { - version: PROTOCOL_VERSION, - cell_width_px: 8, - cell_height_px: 16, - surface_size: crate::protocol::ClientSurfaceSize { cols: 0, rows: 29 }, - pixel_mouse: false, - direct_graphics: false, - endpoint_keybindings: false, - mouse_capture: false, - }, - ) - .expect("write empty shell hello"); + protocol::write_message(&mut client_stream, &endpoint_hello(0, 29)) + .expect("write empty shell hello"); let welcome: ServerMessage = protocol::read_message(&mut client_stream, MAX_FRAME_SIZE).expect("read welcome"); - assert!(matches!( - welcome, - ServerMessage::Welcome { - encoding: RenderEncoding::SemanticFrame, - error: Some(error), - .. - } if error.contains("non-empty pane surface") - )); + let welcome = endpoint_welcome(welcome); + assert!(welcome + .error + .is_some_and(|error| error.message.contains("non-empty pane surface"))); handle .join() .expect("handshake thread join") @@ -1649,6 +1852,72 @@ mod tests { assert!(server_event_rx.try_recv().is_err()); } + #[test] + fn client_read_loop_ignores_unknown_endpoint_control() { + let (mut client_stream, server_stream, _path) = + local_stream_pair("client-read-future-control"); + let (server_event_tx, mut server_event_rx) = mpsc::channel(4); + let should_quit = Arc::new(AtomicBool::new(false)); + let read_quit = should_quit.clone(); + let handle = std::thread::spawn(move || { + client_read_loop(server_stream, 7, &server_event_tx, &read_quit) + }); + + protocol::write_message( + &mut client_stream, + &ClientMessage::EndpointControl { + kind: "future.optional.v1".into(), + data: "{}".into(), + }, + ) + .unwrap(); + protocol::write_message(&mut client_stream, &ClientMessage::Detach).unwrap(); + + assert!(matches!( + recv_server_event(&mut server_event_rx, "detach after future control"), + ServerEvent::ClientDetach { client_id: 7 } + )); + handle + .join() + .expect("read thread join") + .expect("read thread result"); + } + + #[test] + fn client_read_loop_closes_on_unsafe_shell_resize() { + let (mut client_stream, server_stream, _path) = + local_stream_pair("client-read-unsafe-resize"); + let (server_event_tx, mut server_event_rx) = mpsc::channel(4); + let should_quit = Arc::new(AtomicBool::new(false)); + let read_quit = should_quit.clone(); + let handle = std::thread::spawn(move || { + client_read_loop(server_stream, 7, &server_event_tx, &read_quit) + }); + + protocol::write_message( + &mut client_stream, + &ClientMessage::ClientShellResize { + cell_width_px: 8, + cell_height_px: 16, + surface_size: crate::protocol::ClientSurfaceSize { + cols: MAX_CLIENT_SHELL_DIMENSION, + rows: MAX_CLIENT_SHELL_DIMENSION, + }, + pixel_mouse: false, + }, + ) + .unwrap(); + + assert!(matches!( + recv_server_event(&mut server_event_rx, "unsafe resize disconnect"), + ServerEvent::ClientDisconnected { client_id: 7 } + )); + handle + .join() + .expect("read thread join") + .expect("read thread result"); + } + #[test] fn client_read_loop_rejects_oversized_bracketed_paste_without_disconnect() { let (mut client_stream, server_stream, _path) = local_stream_pair("client-read-oversized"); diff --git a/src/server/headless.rs b/src/server/headless.rs index 6bae0a98..1cf8ebd4 100644 --- a/src/server/headless.rs +++ b/src/server/headless.rs @@ -1934,12 +1934,17 @@ impl HeadlessServer { connection.shell_projection_revision, config_diagnostic, ); - connection.shell_snapshot = Some(snapshot.clone()); + let snapshot_message = match crate::protocol::endpoint::snapshot_message(&snapshot) + { + Ok(message) => message, + Err(err) => { + warn!(client_id, err = %err, "failed to encode endpoint snapshot"); + return false; + } + }; + connection.shell_snapshot = Some(snapshot); self.clients.insert(client_id, connection); - self.send_to_client( - client_id, - ServerMessage::ClientShellSnapshot(Box::new(snapshot)), - ); + self.send_to_client(client_id, snapshot_message); self.foreground_client_id = Some(client_id); if first_app_client { self.app.mark_git_status_refresh_due(Instant::now()); @@ -2374,6 +2379,26 @@ impl HeadlessServer { } foreground_changed || runtime.scroll_metrics() != scroll_before } + ServerEvent::ClientShellEndpointRequestError { + client_id, + boot_id, + request_id, + code, + message, + } => { + let Some(client) = self.clients.get(&client_id) else { + return false; + }; + if !matches!(client.mode, ClientConnectionMode::ClientShell) { + self.remove_client_and_resize_if_needed(client_id); + return true; + } + let message = crate::server::client_commands::error_message( + boot_id, request_id, code, message, + ); + self.send_to_client(client_id, message); + false + } ServerEvent::ClientShellEndpointRequest { client_id, boot_id, diff --git a/src/server/headless/render.rs b/src/server/headless/render.rs index cab79de2..19581abc 100644 --- a/src/server/headless/render.rs +++ b/src/server/headless/render.rs @@ -387,11 +387,18 @@ impl HeadlessServer { client.shell_projection_revision = client.shell_projection_revision.saturating_add(1); candidate.revision = client.shell_projection_revision; - let message = ServerMessage::ClientShellSnapshot(Box::new(candidate.clone())); + let message = match crate::protocol::endpoint::snapshot_message(&candidate) { + Ok(message) => message, + Err(err) => { + warn!(client_id, err = %err, "failed to encode endpoint snapshot"); + broken_clients.push(client_id); + continue; + } + }; let framed = match Self::frame_server_message(&message) { Ok(framed) => framed, Err(err) => { - warn!(client_id, err = %err, "failed to frame client shell replacement"); + warn!(client_id, err = %err, "failed to frame endpoint snapshot"); broken_clients.push(client_id); continue; } diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs index f5c33f6a..93c7420c 100644 --- a/src/server/headless/tests/mod.rs +++ b/src/server/headless/tests/mod.rs @@ -3,6 +3,14 @@ use super::*; #[path = "pane_graphics.rs"] mod pane_graphics_tests; +fn client_shell_snapshot(message: ServerMessage) -> Box { + let ServerMessage::EndpointControl { kind, data } = message else { + panic!("expected client shell snapshot"); + }; + assert_eq!(kind, crate::protocol::endpoint::ENDPOINT_SNAPSHOT_KIND); + Box::new(serde_json::from_str(&data).expect("decode client shell snapshot")) +} + fn test_headless_server() -> HeadlessServer { test_headless_server_with_event_hub(api::EventHub::default()) } @@ -671,6 +679,33 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() { shutdown_test_runtimes(&mut server); } +#[test] +fn terminal_client_endpoint_request_error_removes_client() { + let mut server = test_headless_server(); + let (writer, _control_rx, _render_rx) = test_client_writer(); + let client_id = 42; + assert!(!server.handle_server_event(ServerEvent::ClientConnected { + client_id, + cols: 80, + rows: 24, + cell_width_px: 0, + cell_height_px: 0, + pixel_mouse: false, + writer, + })); + + assert!( + server.handle_server_event(ServerEvent::ClientShellEndpointRequestError { + client_id, + boot_id: "boot".into(), + request_id: "request".into(), + code: "unsupported_method", + message: "unsupported".into(), + }) + ); + assert!(!server.clients.contains_key(&client_id)); +} + #[tokio::test] async fn client_shell_receives_metadata_then_shell_free_pane_surface() { let mut server = test_headless_server(); @@ -713,25 +748,23 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() { writer, }) ); - match read_server_message(control_rx.recv().expect("shell snapshot")) { - ServerMessage::ClientShellSnapshot(snapshot) => { - assert_eq!(snapshot.workspaces.len(), 1); - assert_eq!(snapshot.workspaces[0].label, "shell-only-label"); - assert_eq!( - snapshot.config_diagnostic.as_deref(), - Some("endpoint config warning") - ); - assert_eq!( - snapshot.product_announcement.as_ref().map(|announcement| ( - announcement.version.as_str(), - announcement.id.as_str(), - announcement.preview, - )), - Some(("0.8.2", "client-shell", true)) - ); - } - other => panic!("expected client shell snapshot, got {other:?}"), - } + let snapshot = client_shell_snapshot(read_server_message( + control_rx.recv().expect("shell snapshot"), + )); + assert_eq!(snapshot.workspaces.len(), 1); + assert_eq!(snapshot.workspaces[0].label, "shell-only-label"); + assert_eq!( + snapshot.config_diagnostic.as_deref(), + Some("endpoint config warning") + ); + assert_eq!( + snapshot.product_announcement.as_ref().map(|announcement| ( + announcement.version.as_str(), + announcement.id.as_str(), + announcement.preview, + )), + Some(("0.8.2", "client-shell", true)) + ); server.render_and_stream(); let initial_surface = match read_server_message(render_rx.recv().expect("pane surface")) { @@ -1132,11 +1165,9 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() { writer: local_writer, }) ); - let ServerMessage::ClientShellSnapshot(local_snapshot) = - read_server_message(local_control.recv().expect("local shell snapshot")) - else { - panic!("expected local shell snapshot"); - }; + let local_snapshot = client_shell_snapshot(read_server_message( + local_control.recv().expect("local shell snapshot"), + )); assert_eq!( local_snapshot.config_diagnostic.as_deref(), Some("theme warning") @@ -1157,11 +1188,9 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() { writer: endpoint_writer, }) ); - let ServerMessage::ClientShellSnapshot(endpoint_snapshot) = - read_server_message(endpoint_control.recv().expect("endpoint shell snapshot")) - else { - panic!("expected endpoint shell snapshot"); - }; + let endpoint_snapshot = client_shell_snapshot(read_server_message( + endpoint_control.recv().expect("endpoint shell snapshot"), + )); assert_eq!( endpoint_snapshot.config_diagnostic.as_deref(), Some("server keybinding warning\ntheme warning") @@ -1199,10 +1228,10 @@ async fn client_shell_replaces_projection_and_focuses_stable_ids() { writer, }) ); - let initial_revision = match read_server_message(control_rx.recv().expect("initial snapshot")) { - ServerMessage::ClientShellSnapshot(snapshot) => snapshot.revision, - other => panic!("expected initial shell snapshot, got {other:?}"), - }; + let initial_revision = client_shell_snapshot(read_server_message( + control_rx.recv().expect("initial snapshot"), + )) + .revision; let _ = server.app.handle_api_request(crate::api::schema::Request { id: "test.client.shell.workspace.focus".into(), @@ -1213,10 +1242,9 @@ async fn client_shell_replaces_projection_and_focuses_stable_ids() { assert_eq!(server.app.state.active, Some(1)); server.render_and_stream(); - let replacement = match read_server_message(control_rx.recv().expect("replacement snapshot")) { - ServerMessage::ClientShellSnapshot(snapshot) => snapshot, - other => panic!("expected replacement shell snapshot, got {other:?}"), - }; + let replacement = client_shell_snapshot(read_server_message( + control_rx.recv().expect("replacement snapshot"), + )); assert!(replacement.revision > initial_revision); assert_eq!( replacement.focused_workspace_id.as_deref(), @@ -1441,9 +1469,8 @@ async fn client_shell_streams_and_targets_popup_terminal_content() { writer, }) ); - assert!(matches!( - read_server_message(control_rx.recv().expect("shell snapshot")), - ServerMessage::ClientShellSnapshot(_) + let _snapshot = client_shell_snapshot(read_server_message( + control_rx.recv().expect("shell snapshot"), )); server.render_and_stream(); @@ -4553,7 +4580,7 @@ fn update_notification_is_semantic_for_system_delivery() { assert_eq!(notification.title, "Herdr v9.9.9 available"); assert_eq!( notification.body.as_deref(), - Some("detach, run `herdr update`, then follow its restart guidance") + Some("detach, run `herdr update`, then run Herdr again to reconnect") ); } other => panic!("expected semantic update notification, got {other:?}"), diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs index e533bd03..15b23bf8 100644 --- a/src/server/render_scale_benchmark.rs +++ b/src/server/render_scale_benchmark.rs @@ -16,6 +16,7 @@ const ROWS: u16 = 40; const SAMPLE_COUNT: usize = 40; const WARMUP_COUNT: usize = 5; const CARDINALITIES: [usize; 3] = [1, 15, 50]; +const CLIENT_CARDINALITIES: [usize; 2] = [1, 4]; #[derive(Clone, Copy)] struct StageStats { @@ -210,9 +211,52 @@ fn print_profiles(label: &str, build: fn(usize) -> Vec) { print_stage("combined pipeline", &rows, |stats| stats.total); } +fn profile_snapshot_encoding( + build: fn(usize) -> Vec, + count: usize, + client_count: usize, +) -> StageStats { + let pipeline = RenderPipeline::new(build(count)); + let run = || { + let started = Instant::now(); + let template = super::client_shell::snapshot(&pipeline.app, "bench-boot", 1, None); + for client_index in 0..client_count { + let mut snapshot = template.clone(); + snapshot.revision = client_index as u64 + 1; + let message = crate::protocol::endpoint::snapshot_message(&snapshot) + .expect("benchmark snapshot should serialize"); + black_box( + bincode::serde::encode_to_vec(message, bincode::config::standard()) + .expect("benchmark snapshot message should frame"), + ); + } + started.elapsed() + }; + for _ in 0..WARMUP_COUNT { + black_box(run()); + } + summarize((0..SAMPLE_COUNT).map(|_| run()).collect()) +} + +fn print_snapshot_encoding_profiles(label: &str, build: fn(usize) -> Vec) { + println!("{label} snapshot projection + JSON framing"); + println!(" panes clients median_us p95_us max_us"); + for count in CARDINALITIES { + for client_count in CLIENT_CARDINALITIES { + let stats = profile_snapshot_encoding(build, count, client_count); + println!( + " {count:>10} {client_count:>7} {:>9} {:>6} {:>6}", + stats.median_us, stats.p95_us, stats.max_us + ); + } + } +} + #[tokio::test(flavor = "current_thread")] #[ignore = "manual client-rendered pipeline scaling profile"] async fn render_scale_profile() { print_profiles("background workspaces (one pane each)", workspaces); + print_snapshot_encoding_profiles("background workspaces", workspaces); print_profiles("active panes (one workspace)", active_panes); + print_snapshot_encoding_profiles("active panes", active_panes); } -- cgit v1.2.3