diff options
Diffstat (limited to 'src/server')
| -rw-r--r-- | src/server/client_transport.rs | 129 | ||||
| -rw-r--r-- | src/server/headless/render.rs | 10 | ||||
| -rw-r--r-- | src/server/headless/tests/mod.rs | 47 |
3 files changed, 183 insertions, 3 deletions
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs index c57d0f86..da682c06 100644 --- a/src/server/client_transport.rs +++ b/src/server/client_transport.rs @@ -41,6 +41,9 @@ const MIN_CLIENT_ROWS: u16 = 1; /// and cleanup overhead. const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(4); +#[cfg(unix)] +const OBSERVER_WRITE_TIMEOUT: Duration = Duration::from_secs(30); + /// 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; @@ -940,7 +943,11 @@ fn client_writer_loop( } fn write_framed_bytes(stream: &mut LocalStream, data: &[u8]) -> bool { - if let Err(err) = stream.write_all(data) { + #[cfg(unix)] + let result = crate::platform::write_client_stream(stream, data); + #[cfg(windows)] + let result = stream.write_all(data); + if let Err(err) = result { debug!(err = %err, "client write failed, closing writer"); return false; } @@ -970,8 +977,14 @@ fn client_read_loop_with_endpoint_controls( endpoint_control_writer: Option<&ClientControlWriter>, ) -> io::Result<()> { while !should_quit.load(Ordering::Acquire) { - let msg: ClientMessage = match protocol::read_message(&mut stream, MAX_GRAPHICS_FRAME_SIZE) - { + #[cfg(unix)] + let message = protocol::read_message( + &mut crate::platform::ClientStreamReader(&mut stream), + MAX_GRAPHICS_FRAME_SIZE, + ); + #[cfg(windows)] + let message = protocol::read_message(&mut stream, MAX_GRAPHICS_FRAME_SIZE); + let msg: ClientMessage = match message { Ok(msg) => msg, Err(protocol::FramingError::UnexpectedEof) => { // Client disconnected. @@ -1027,6 +1040,13 @@ fn client_read_loop_with_endpoint_controls( } } ClientMessage::ObserveTerminal { target } => { + #[cfg(unix)] + { + stream.set_send_timeout(Some(OBSERVER_WRITE_TIMEOUT))?; + // macOS Unix sockets can block even with per-send MSG_DONTWAIT. + // ClientStreamReader preserves blocking reads on the shared socket. + stream.set_nonblocking(true)?; + } ServerEvent::ClientObserveTerminal { client_id, target } } ClientMessage::ControlTerminal { target, takeover } => { @@ -1641,6 +1661,109 @@ mod tests { )); } + #[cfg(unix)] + #[test] + fn observer_write_timeout_resets_when_sending_makes_progress() { + use std::io::Read as _; + + let (mut client, mut server, _path) = local_stream_pair("slow-observer"); + server + .set_send_timeout(Some(Duration::from_millis(100))) + .unwrap(); + server.set_nonblocking(true).unwrap(); + let worker = std::thread::spawn(move || { + assert!(write_framed_bytes(&mut server, &vec![b'x'; 1024 * 1024])); + }); + client + .set_recv_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let mut received = 0; + let mut buffer = [0; 16 * 1024]; + while received < 1024 * 1024 { + let count = client.read(&mut buffer).unwrap(); + assert_ne!(count, 0, "observer disconnected while making progress"); + received += count; + std::thread::sleep(Duration::from_millis(5)); + } + worker.join().unwrap(); + } + + #[cfg(unix)] + #[test] + fn stalled_observer_timeout_releases_writer_and_reader() { + let (mut client, server, _path) = local_stream_pair("stalled-observer"); + let writer_stream = server.try_clone().expect("clone writer stream"); + let (writer, queue) = test_queue_writer(); + let (events, event_rx) = mpsc::channel(8); + let reader_events = events.clone(); + let (reader_done_tx, reader_done) = std::sync::mpsc::channel(); + let reader = std::thread::spawn(move || { + let result = client_read_loop( + server, + 14, + &reader_events, + &Arc::new(AtomicBool::new(false)), + ); + let _ = reader_done_tx.send(result); + }); + protocol::write_message( + &mut client, + &ClientMessage::ObserveTerminal { + target: "w1:p1".into(), + }, + ) + .expect("observe request"); + let mut event_rx = event_rx; + assert!(matches!( + event_rx.blocking_recv(), + Some(ServerEvent::ClientObserveTerminal { client_id: 14, .. }) + )); + let LocalStream::UdSocket(socket) = &writer_stream; + assert_eq!( + socket.inner().write_timeout().unwrap(), + Some(Duration::from_secs(30)) + ); + assert_eq!(socket.inner().read_timeout().unwrap(), None); + let mut resize = Vec::new(); + protocol::write_message( + &mut resize, + &ClientMessage::Resize { + cols: 80, + rows: 24, + cell_width_px: 0, + cell_height_px: 0, + pixel_mouse: false, + }, + ) + .unwrap(); + client.write_all(&resize[..2]).unwrap(); + std::thread::sleep(Duration::from_millis(20)); + client.write_all(&resize[2..]).unwrap(); + assert!(matches!( + event_rx.blocking_recv(), + Some(ServerEvent::ClientResize { client_id: 14, .. }) + )); + writer_stream + .set_send_timeout(Some(Duration::from_millis(200))) + .unwrap(); + let (writer_done_tx, writer_done) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + client_writer_loop(writer_stream, 14, queue, events); + let _ = writer_done_tx.send(()); + }); + writer.render.try_send(vec![0; 4 * 1024 * 1024]).unwrap(); + writer_done + .recv_timeout(Duration::from_millis(350)) + .expect("writer timed out"); + reader_done + .recv_timeout(Duration::from_secs(2)) + .expect("reader released") + .unwrap(); + worker.join().unwrap(); + reader.join().unwrap(); + assert!(writer.control.send(vec![1]).is_err()); + } + #[test] fn clamp_terminal_size_zero_zero() { assert_eq!( diff --git a/src/server/headless/render.rs b/src/server/headless/render.rs index fc55417f..882a1615 100644 --- a/src/server/headless/render.rs +++ b/src/server/headless/render.rs @@ -462,6 +462,16 @@ impl HeadlessServer { let mut broken_clients: Vec<u64> = Vec::new(); for (client_id, (cols, rows), cell_size, _is_foreground, mode) in render_targets { + #[cfg(unix)] + if matches!(mode, ClientConnectionMode::TerminalObserve { .. }) + && self + .clients + .get(&client_id) + .is_some_and(|client| client.deferred_render() != DeferredRender::None) + { + // The writer-drained event schedules recovery, even if pane output stops. + continue; + } let area = Rect::new(0, 0, cols, rows); let shell_target = self.shell_target_for_client(client_id); let shell_tab_id = self.shell_tab_id_for_client(client_id); diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs index 2c13dc8b..bb196be5 100644 --- a/src/server/headless/tests/mod.rs +++ b/src/server/headless/tests/mod.rs @@ -3509,6 +3509,53 @@ fn terminal_attach_disconnect_restores_client_shell_pane_size() { rt.shutdown_timeout(Duration::from_millis(100)); } +#[cfg(unix)] +#[test] +fn backpressured_observer_skips_runtime_access_and_recovers_without_new_output() { + with_terminal_session_test_server(|server, terminal_id, target, _| { + let (writer, control, frames) = test_client_writer(); + server.handle_server_event(ServerEvent::ClientConnected { + client_id: 7, + cols: 80, + rows: 24, + cell_width_px: 0, + cell_height_px: 0, + pixel_mouse: false, + writer, + }); + server.handle_server_event(ServerEvent::ClientObserveTerminal { + client_id: 7, + target, + }); + server.render_and_stream(); + server.app.terminal_runtimes.insert( + terminal_id.clone(), + crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 24, b"LATEST"), + ); + server.render_and_stream(); + assert_eq!(server.clients[&7].deferred_render(), DeferredRender::Full); + + // An absent runtime makes any attempted rendering observable without timing a lock. + let runtime = server.app.terminal_runtimes.remove(&terminal_id).unwrap(); + server.render_and_stream(); + server.app.terminal_runtimes.insert(terminal_id, runtime); + assert!( + server.clients.contains_key(&7), + "backpressure must skip runtime access" + ); + assert!(control.try_recv().is_err()); + + let _ = frames.recv().expect("previously accepted frame"); + assert!(server.handle_server_event(ServerEvent::ClientWriterDrained { client_id: 7 })); + server.render_and_stream(); + let ServerMessage::Terminal(frame) = read_server_message(frames.recv().unwrap()) else { + panic!("terminal update"); + }; + assert!(String::from_utf8_lossy(&frame.bytes).contains("LATEST")); + assert_eq!(server.clients[&7].deferred_render(), DeferredRender::None); + }); +} + #[test] fn terminal_observe_allows_multiple_clients_without_attach_ownership() { with_terminal_session_test_server(|server, terminal_id, terminal_id_string, _| { |