From d4a5e02e8433db156b9dabe2b81ddd1d2aca9ea6 Mon Sep 17 00:00:00 2001 From: Ogulcan Celik Date: Mon, 25 May 2026 15:36:50 +0300 Subject: feat: hand off ptys during updates --- src/server/client_accept.rs | 19 ++ src/server/handoff.rs | 446 ++++++++++++++++++++++++++++++++++++++ src/server/headless.rs | 509 +++++++++++++++++++++++++++++++++++++++++++- src/server/mod.rs | 2 + 4 files changed, 970 insertions(+), 6 deletions(-) create mode 100644 src/server/handoff.rs (limited to 'src/server') diff --git a/src/server/client_accept.rs b/src/server/client_accept.rs index 61db2eb9..afb2c398 100644 --- a/src/server/client_accept.rs +++ b/src/server/client_accept.rs @@ -48,3 +48,22 @@ pub(crate) fn accept_pending_client_connections( Ok(()) } + +/// Drains pending thin-client connections without starting handshakes. +/// +/// During live handoff the old server must not let clients sit in the Unix +/// listener backlog waiting for a welcome frame that will never be sent. +pub(crate) fn reject_pending_client_connections(listener: &UnixListener) -> io::Result<()> { + loop { + match listener.accept() { + Ok((_stream, _addr)) => {} + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, + Err(err) => { + error!(err = %err, "client listener reject failed"); + break; + } + } + } + + Ok(()) +} diff --git a/src/server/handoff.rs b/src/server/handoff.rs new file mode 100644 index 00000000..d38bdc29 --- /dev/null +++ b/src/server/handoff.rs @@ -0,0 +1,446 @@ +#[cfg(unix)] +use std::io::{self, Read, Write}; +#[cfg(unix)] +use std::os::fd::{AsRawFd, RawFd}; +#[cfg(unix)] +use std::os::unix::net::{UnixListener, UnixStream}; +#[cfg(unix)] +use std::os::unix::process::CommandExt; +#[cfg(unix)] +use std::path::{Path, PathBuf}; +#[cfg(unix)] +use std::process::Command; +#[cfg(unix)] +use std::time::Duration; + +#[cfg(unix)] +use serde::{Deserialize, Serialize}; +#[cfg(unix)] +use tracing::{info, warn}; + +#[cfg(unix)] +const HANDOFF_VERSION: u32 = 1; +#[cfg(unix)] +const READY_TIMEOUT: Duration = Duration::from_secs(30); +#[cfg(unix)] +pub(crate) const MAX_FDS_PER_HANDOFF: usize = 64; +#[cfg(unix)] +pub(crate) const MAX_REPLAY_BYTES_PER_PANE: usize = 8 * 1024; +#[cfg(unix)] +pub(crate) const COMMIT_TIMEOUT: Duration = READY_TIMEOUT; + +#[cfg(unix)] +#[derive(Serialize, Deserialize)] +pub(crate) struct HandoffManifest { + pub version: u32, + pub source_version: String, + pub source_protocol: u32, + pub expected_version: Option, + pub expected_protocol: Option, + pub snapshot: crate::persist::SessionSnapshot, + pub panes: Vec, +} + +#[cfg(unix)] +#[derive(Debug, Serialize, Deserialize)] +pub(crate) struct HandoffPane { + pub pane_id: u32, + pub child_pid: u32, + pub rows: u16, + pub cols: u16, + pub cell_width_px: u32, + pub cell_height_px: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub initial_history_ansi: Option, +} + +#[cfg(unix)] +pub(crate) struct ReceivedHandoff { + pub manifest: HandoffManifest, + pub fds: Vec, + pub stream: UnixStream, +} + +#[cfg(unix)] +pub(crate) fn handoff_socket_path() -> PathBuf { + crate::session::data_dir().join(format!("herdr-handoff-{}.sock", std::process::id())) +} + +#[cfg(unix)] +pub(crate) fn spawn_handoff_import( + import_exe: Option<&Path>, + socket_path: &Path, + token: &str, +) -> io::Result { + let fallback_exe; + let exe = if let Some(import_exe) = import_exe { + import_exe + } else { + fallback_exe = std::env::current_exe().map_err(|err| { + io::Error::new( + err.kind(), + format!("failed to determine herdr executable path: {err}"), + ) + })?; + &fallback_exe + }; + let mut command = Command::new(exe); + command + .arg("server") + .arg("--handoff-import") + .arg(socket_path) + .arg(token) + .process_group(0) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()); + let child = command.spawn().map_err(|err| { + io::Error::new( + err.kind(), + format!( + "failed to spawn handoff import server at {}: {err}", + exe.display() + ), + ) + })?; + Ok(child.id()) +} + +#[cfg(unix)] +pub(crate) fn bind_listener(socket_path: &Path) -> io::Result { + let _ = std::fs::remove_file(socket_path); + let listener = UnixListener::bind(socket_path)?; + listener.set_nonblocking(true)?; + restrict_socket_permissions(socket_path)?; + Ok(listener) +} + +#[cfg(unix)] +pub(crate) fn accept_and_validate_on( + listener: UnixListener, + socket_path: &Path, + token: &str, + manifest: &HandoffManifest, +) -> io::Result { + let (mut stream, _) = accept_with_timeout(&listener, READY_TIMEOUT)?; + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(READY_TIMEOUT))?; + stream.set_write_timeout(Some(READY_TIMEOUT))?; + let token_line = read_line_unbuffered(&mut stream)?; + if token_line.trim_end() != token { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "handoff import token mismatch", + )); + } + + serde_json::to_writer(&mut stream, manifest).map_err(io::Error::other)?; + stream.write_all(b"\n")?; + stream.flush()?; + + stream.set_read_timeout(Some(READY_TIMEOUT))?; + let validated = read_line_unbuffered(&mut stream)?; + if validated.trim_end() != "validated" { + return Err(io::Error::other("handoff import did not validate manifest")); + } + let _ = std::fs::remove_file(socket_path); + Ok(stream) +} + +#[cfg(unix)] +pub(crate) fn send_fds_and_wait_restored(stream: &mut UnixStream, fds: &[RawFd]) -> io::Result<()> { + if fds.len() > MAX_FDS_PER_HANDOFF { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("handoff supports at most {MAX_FDS_PER_HANDOFF} pane file descriptors at once"), + )); + } + send_fds(stream, fds)?; + + stream.set_read_timeout(Some(READY_TIMEOUT))?; + let restored = read_line_unbuffered(&mut *stream)?; + if restored.trim_end() != "restored" { + return Err(io::Error::other( + "handoff import did not report restored runtimes", + )); + } + Ok(()) +} + +#[cfg(unix)] +pub(crate) fn wait_ready(stream: &mut UnixStream) -> io::Result<()> { + stream.set_read_timeout(Some(READY_TIMEOUT))?; + let ready = read_line_unbuffered(&mut *stream)?; + if ready.trim_end() != "ready" { + return Err(io::Error::other("handoff import did not report ready")); + } + Ok(()) +} + +#[cfg(unix)] +pub(crate) fn report_committed(stream: &mut UnixStream) -> io::Result<()> { + stream.write_all(b"committed\n")?; + stream.flush() +} + +#[cfg(unix)] +pub(crate) fn wait_owned_ack(stream: &mut UnixStream) { + if let Err(err) = stream.set_read_timeout(Some(READY_TIMEOUT)) { + warn!(err = %err, "failed to set handoff ownership ack timeout"); + return; + } + match read_line_unbuffered(&mut *stream) { + Ok(owned) if owned.trim_end() == "owned" => {} + Ok(other) => { + warn!( + response = %other.trim_end(), + "handoff import sent unexpected ownership ack after commit" + ); + } + Err(err) => { + warn!(err = %err, "handoff import ownership ack was not received after commit"); + } + } +} + +#[cfg(unix)] +pub(crate) fn receive(socket_path: &Path, token: &str) -> io::Result { + let mut stream = UnixStream::connect(socket_path)?; + stream.write_all(token.as_bytes())?; + stream.write_all(b"\n")?; + stream.flush()?; + + let manifest_line = read_line_unbuffered(&mut stream)?; + let manifest: HandoffManifest = + serde_json::from_str(&manifest_line).map_err(io::Error::other)?; + if manifest.version != HANDOFF_VERSION { + return Err(io::Error::other(format!( + "unsupported handoff version {}", + manifest.version + ))); + } + if manifest + .expected_protocol + .is_some_and(|protocol| protocol != crate::protocol::PROTOCOL_VERSION) + { + return Err(io::Error::other(format!( + "handoff expected protocol {}, but this server speaks protocol {}", + manifest.expected_protocol.unwrap_or_default(), + crate::protocol::PROTOCOL_VERSION + ))); + } + if manifest + .expected_version + .as_deref() + .is_some_and(|version| version != env!("CARGO_PKG_VERSION")) + { + return Err(io::Error::other(format!( + "handoff expected herdr v{}, but this server is v{}", + manifest.expected_version.as_deref().unwrap_or("unknown"), + env!("CARGO_PKG_VERSION") + ))); + } + stream.write_all(b"validated\n")?; + stream.flush()?; + let fds = recv_fds(&stream, manifest.panes.len())?; + Ok(ReceivedHandoff { + manifest, + fds, + stream, + }) +} + +#[cfg(unix)] +pub(crate) fn report_restored(stream: &mut UnixStream) -> io::Result<()> { + stream.write_all(b"restored\n")?; + stream.flush() +} + +#[cfg(unix)] +pub(crate) fn report_ready(stream: &mut UnixStream) -> io::Result<()> { + stream.write_all(b"ready\n")?; + stream.flush() +} + +#[cfg(unix)] +pub(crate) fn wait_committed(stream: &mut UnixStream) -> io::Result<()> { + stream.set_read_timeout(Some(READY_TIMEOUT))?; + let committed = read_line_unbuffered(&mut *stream)?; + if committed.trim_end() != "committed" { + return Err(io::Error::other("handoff source did not commit")); + } + Ok(()) +} + +#[cfg(unix)] +pub(crate) fn report_owned(stream: &mut UnixStream) -> io::Result<()> { + stream.write_all(b"owned\n")?; + stream.flush() +} + +#[cfg(unix)] +pub(crate) fn manifest_for( + snapshot: crate::persist::SessionSnapshot, + panes: Vec, + expected_protocol: Option, + expected_version: Option, +) -> HandoffManifest { + HandoffManifest { + version: HANDOFF_VERSION, + source_version: env!("CARGO_PKG_VERSION").to_string(), + source_protocol: crate::protocol::PROTOCOL_VERSION, + expected_version, + expected_protocol, + snapshot, + panes, + } +} + +#[cfg(unix)] +fn restrict_socket_permissions(path: &Path) -> io::Result<()> { + use std::os::unix::fs::PermissionsExt; + + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)) +} + +#[cfg(unix)] +fn accept_with_timeout( + listener: &UnixListener, + timeout: Duration, +) -> io::Result<(UnixStream, std::os::unix::net::SocketAddr)> { + let deadline = std::time::Instant::now() + timeout; + loop { + match listener.accept() { + Ok(accepted) => return Ok(accepted), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => { + if std::time::Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "timed out waiting for handoff import connection", + )); + } + std::thread::sleep(Duration::from_millis(25)); + } + Err(err) if err.kind() == io::ErrorKind::Interrupted => {} + Err(err) => return Err(err), + } + } +} + +#[cfg(unix)] +fn read_line_unbuffered(stream: &mut UnixStream) -> io::Result { + let mut bytes = Vec::new(); + let mut byte = [0u8; 1]; + loop { + let read = stream.read(&mut byte)?; + if read == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "handoff stream closed while reading line", + )); + } + bytes.push(byte[0]); + if byte[0] == b'\n' { + return String::from_utf8(bytes) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err)); + } + if bytes.len() > 16 * 1024 * 1024 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "handoff line exceeded maximum size", + )); + } + } +} + +#[cfg(unix)] +fn send_fds(stream: &UnixStream, fds: &[RawFd]) -> io::Result<()> { + if fds.is_empty() { + return Ok(()); + } + let byte = [b'F']; + let iov = [libc::iovec { + iov_base: byte.as_ptr() as *mut libc::c_void, + iov_len: byte.len(), + }]; + let fd_bytes = std::mem::size_of_val(fds); + let mut control = vec![0u8; unsafe { libc::CMSG_SPACE(fd_bytes as u32) as usize }]; + let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; + msg.msg_iov = iov.as_ptr() as *mut libc::iovec; + msg.msg_iovlen = iov.len() as _; + msg.msg_control = control.as_mut_ptr() as *mut libc::c_void; + msg.msg_controllen = control.len() as _; + + unsafe { + let cmsg = libc::CMSG_FIRSTHDR(&msg); + if cmsg.is_null() { + return Err(io::Error::other("failed to allocate fd control message")); + } + (*cmsg).cmsg_level = libc::SOL_SOCKET; + (*cmsg).cmsg_type = libc::SCM_RIGHTS; + (*cmsg).cmsg_len = libc::CMSG_LEN(fd_bytes as u32) as _; + std::ptr::copy_nonoverlapping(fds.as_ptr() as *const u8, libc::CMSG_DATA(cmsg), fd_bytes); + if libc::sendmsg(stream.as_raw_fd(), &msg, 0) < 0 { + return Err(io::Error::last_os_error()); + } + } + Ok(()) +} + +#[cfg(unix)] +fn recv_fds(stream: &UnixStream, expected: usize) -> io::Result> { + if expected == 0 { + return Ok(Vec::new()); + } + let mut byte = [0u8; 1]; + let mut iov = [libc::iovec { + iov_base: byte.as_mut_ptr() as *mut libc::c_void, + iov_len: byte.len(), + }]; + let fd_bytes = expected * std::mem::size_of::(); + let mut control = vec![0u8; unsafe { libc::CMSG_SPACE(fd_bytes as u32) as usize }]; + let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; + msg.msg_iov = iov.as_mut_ptr(); + msg.msg_iovlen = iov.len() as _; + msg.msg_control = control.as_mut_ptr() as *mut libc::c_void; + msg.msg_controllen = control.len() as _; + + let read = unsafe { libc::recvmsg(stream.as_raw_fd(), &mut msg, 0) }; + if read < 0 { + return Err(io::Error::last_os_error()); + } + if msg.msg_flags & libc::MSG_CTRUNC != 0 { + return Err(io::Error::other("handoff fd control message was truncated")); + } + + let mut out = Vec::new(); + unsafe { + let cmsg = libc::CMSG_FIRSTHDR(&msg); + if cmsg.is_null() + || (*cmsg).cmsg_level != libc::SOL_SOCKET + || (*cmsg).cmsg_type != libc::SCM_RIGHTS + { + return Err(io::Error::other("handoff fd message missing SCM_RIGHTS")); + } + let data_len = ((*cmsg).cmsg_len as usize).saturating_sub(libc::CMSG_LEN(0) as usize); + let count = data_len / std::mem::size_of::(); + let data = libc::CMSG_DATA(cmsg) as *const RawFd; + for idx in 0..count { + out.push(*data.add(idx)); + } + } + if out.len() != expected { + for fd in out { + let _ = unsafe { libc::close(fd) }; + } + return Err(io::Error::other(format!( + "expected {expected} handoff fds, received fewer" + ))); + } + Ok(out) +} + +#[cfg(unix)] +pub(crate) fn log_import_result(panes: usize) { + info!(panes, "handoff import ready"); +} diff --git a/src/server/headless.rs b/src/server/headless.rs index d10aef04..4dafb6d7 100644 --- a/src/server/headless.rs +++ b/src/server/headless.rs @@ -15,7 +15,6 @@ //! and pane spawn failure during restore use std::collections::HashMap; -use std::fs; use std::io; use std::os::unix::net::UnixListener; use std::path::{Path, PathBuf}; @@ -35,11 +34,14 @@ use crate::api; use crate::app; use crate::config; use crate::events::AppEvent; +use crate::ipc::{remove_socket_file_if_owned, socket_file_identity, SocketFileIdentity}; use crate::protocol::{ self, AttachScrollDirection, AttachScrollSource, FrameData, ServerMessage, MAX_FRAME_SIZE, MAX_GRAPHICS_FRAME_SIZE, }; -use crate::server::client_accept::accept_pending_client_connections; +use crate::server::client_accept::{ + accept_pending_client_connections, reject_pending_client_connections, +}; use crate::server::client_transport::ServerEvent; use crate::server::clients::{ events_include_interaction, latest_app_client, render_targets, terminal_attach_client_ids, @@ -58,6 +60,8 @@ use crate::server::terminal_attach::paste_payload_for_runtime; use crate::protocol::RenderEncoding; #[cfg(test)] use crate::server::client_transport::ClientWriter; +#[cfg(test)] +use std::fs; // --------------------------------------------------------------------------- // Loop event enum for the headless server event loop @@ -100,8 +104,11 @@ const CLIENT_ACCEPT_POLL_INTERVAL: Duration = Duration::from_millis(250); /// The headless server — runs the herdr event loop without a real terminal. pub struct HeadlessServer { app: app::App, + api_tx: Option, + api_server: Option, client_listener: UnixListener, client_socket_path: PathBuf, + client_socket_identity: SocketFileIdentity, clients: HashMap, next_client_id: u64, /// The client currently driving the shared pane runtime size, theme, and input keybindings. @@ -121,6 +128,10 @@ pub struct HeadlessServer { effective_size: (u16, u16), /// Flag set when shutdown is initiated. shutting_down: bool, + /// Flag set while exporting live PTYs to a replacement server. + handoff_in_progress: bool, + /// Imported panes get one app-safe resize nudge after the first client attaches. + pending_handoff_repaint_nudge: bool, /// Flag set by Ctrl+C or `server stop` signal. should_quit: Arc, /// Channel for receiving server events from client connection threads. @@ -209,12 +220,18 @@ impl HeadlessServer { /// 1. Prepares the client socket path (cleans up stale sockets) /// 2. Binds the client socket listener /// 3. Returns the server ready to run - pub fn new(app: app::App, config_diagnostics: &[String]) -> io::Result { + pub fn new( + app: app::App, + config_diagnostics: &[String], + api_tx: Option, + api_server: Option, + ) -> io::Result { let client_path = client_socket_path(); prepare_socket_path(&client_path)?; let listener = UnixListener::bind(&client_path)?; restrict_socket_permissions(&client_path)?; + let client_socket_identity = socket_file_identity(&client_path)?; info!(path = %client_path.display(), "client protocol socket listening"); // Set non-blocking on the listener so we can poll it from the event loop. @@ -230,8 +247,11 @@ impl HeadlessServer { Ok(Self { app, + api_tx, + api_server, client_listener: listener, client_socket_path: client_path, + client_socket_identity, clients: HashMap::new(), next_client_id: 1, foreground_client_id: None, @@ -242,6 +262,8 @@ impl HeadlessServer { next_activity_stamp: 1, effective_size: (MIN_COLS, MIN_ROWS), shutting_down: false, + handoff_in_progress: false, + pending_handoff_repaint_nudge: false, should_quit, server_event_rx, server_event_tx, @@ -539,6 +561,217 @@ impl HeadlessServer { } } + #[cfg(unix)] + fn perform_live_handoff( + &mut self, + params: crate::api::schema::ServerLiveHandoffParams, + ) -> io::Result<()> { + info!("starting live handoff"); + let import_exe = params.import_exe.as_deref().map(std::path::PathBuf::from); + let socket_path = crate::server::handoff::handoff_socket_path(); + let token = format!( + "{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ); + let listener = match crate::server::handoff::bind_listener(&socket_path) { + Ok(listener) => listener, + Err(err) => { + self.handoff_in_progress = false; + return Err(err); + } + }; + + let mut pane_by_terminal = HashMap::new(); + for ws in &self.app.state.workspaces { + for tab in &ws.tabs { + for (pane_id, pane) in &tab.panes { + pane_by_terminal.insert(pane.attached_terminal_id.clone(), pane_id.raw()); + } + } + } + if pane_by_terminal.len() > crate::server::handoff::MAX_FDS_PER_HANDOFF { + let _ = std::fs::remove_file(&socket_path); + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!( + "live handoff supports at most {} panes in one update; close panes or restart herdr normally", + crate::server::handoff::MAX_FDS_PER_HANDOFF + ), + )); + } + + self.handoff_in_progress = true; + self.disconnect_all_clients_for_handoff(); + let _ = reject_pending_client_connections(&self.client_listener); + + let mut paused_terminal_ids = Vec::new(); + for terminal_id in pane_by_terminal.keys() { + if let Some(runtime) = self.app.terminal_runtimes.get(terminal_id) { + if let Err(err) = runtime.pause_handoff_reader(Duration::from_secs(2)) { + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + return Err(err); + } + paused_terminal_ids.push(terminal_id.clone()); + } + } + + let snapshot = crate::persist::capture( + &self.app.state.workspaces, + &self.app.state.terminals, + &self.app.terminal_runtimes, + self.app.state.active, + self.app.state.selected, + self.app.state.agent_panel_scope, + self.app.state.sidebar_width, + self.app.state.sidebar_section_split, + self.app.state.collapsed_space_keys.clone(), + ); + + let mut panes = Vec::new(); + for (terminal_id, runtime) in self.app.terminal_runtimes.iter() { + let Some(pane_id) = pane_by_terminal.get(terminal_id).copied() else { + continue; + }; + let mut handoff_pane = runtime.handoff_pane(pane_id); + let has_agent_session = self + .app + .state + .terminals + .get(terminal_id) + .is_some_and(|terminal| terminal.persisted_agent_session.is_some()); + if !has_agent_session { + handoff_pane.initial_history_ansi = runtime.handoff_history_ansi(); + } + panes.push(handoff_pane); + } + + let manifest = crate::server::handoff::manifest_for( + snapshot, + panes, + params.expected_protocol, + params.expected_version, + ); + let child_pid = match crate::server::handoff::spawn_handoff_import( + import_exe.as_deref(), + &socket_path, + &token, + ) { + Ok(child_pid) => child_pid, + Err(err) => { + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + return Err(err); + } + }; + info!(pid = child_pid, socket = %socket_path.display(), "spawned handoff import server"); + + let mut fds = Vec::new(); + let duplicate_result = (|| { + for (terminal_id, runtime) in self.app.terminal_runtimes.iter() { + if !pane_by_terminal.contains_key(terminal_id) { + continue; + } + fds.push(runtime.duplicate_handoff_fd()?); + } + Ok::<(), io::Error>(()) + })(); + if let Err(err) = duplicate_result { + for fd in fds { + let _ = unsafe { libc::close(fd) }; + } + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + return Err(err); + } + + let mut stream = match crate::server::handoff::accept_and_validate_on( + listener, + &socket_path, + &token, + &manifest, + ) { + Ok(stream) => stream, + Err(err) => { + for fd in fds { + let _ = unsafe { libc::close(fd) }; + } + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + return Err(err); + } + }; + + let send_result = crate::server::handoff::send_fds_and_wait_restored(&mut stream, &fds); + for fd in fds { + let _ = unsafe { libc::close(fd) }; + } + if let Err(err) = send_result { + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + return Err(err); + } + + if let Some(api_server) = &self.api_server { + let _ = api_server.remove_socket_file_if_owned(); + } else { + let _ = std::fs::remove_file(crate::api::socket_path()); + } + let _ = remove_socket_file_if_owned(&self.client_socket_path, self.client_socket_identity); + if let Err(err) = crate::server::handoff::wait_ready(&mut stream) { + match self.wait_then_restore_public_sockets_after_failed_handoff() { + Ok(()) => { + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + } + Err(restore_err) => { + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + return Err(io::Error::other(format!( + "handoff replacement server did not become ready: {err}; old server could not restore public sockets: {restore_err}" + ))); + } + } + return Err(io::Error::other(format!( + "handoff replacement server did not become ready: {err}" + ))); + } + if let Err(err) = crate::server::handoff::report_committed(&mut stream) { + match self.wait_then_restore_public_sockets_after_failed_handoff() { + Ok(()) => { + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + } + Err(restore_err) => { + self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); + return Err(io::Error::other(format!( + "handoff replacement server was ready, but commit failed: {err}; old server could not restore public sockets: {restore_err}" + ))); + } + } + return Err(err); + } + + for (terminal_id, runtime) in self.app.terminal_runtimes.drain_for_handoff() { + if !pane_by_terminal.contains_key(&terminal_id) { + continue; + } + debug!(terminal = %terminal_id, "preserving pane runtime for handoff"); + runtime.preserve_for_handoff(); + } + crate::server::handoff::wait_owned_ack(&mut stream); + + self.shutting_down = true; + self.app.state.should_quit = true; + self.app.no_session = true; + info!("live handoff completed; old server exiting"); + Ok(()) + } + + #[cfg(not(unix))] + fn perform_live_handoff( + &mut self, + _params: crate::api::schema::ServerLiveHandoffParams, + ) -> io::Result<()> { + Err(io::Error::other("live handoff is only supported on Unix")) + } + fn sync_visible_server_config_diagnostic(&mut self, uses_local_keybindings: bool) { let visible = if uses_local_keybindings { &self.server_config_diagnostic_without_keybindings @@ -552,6 +785,64 @@ impl HeadlessServer { } } + #[cfg(unix)] + fn restore_public_sockets_after_failed_handoff(&mut self) -> io::Result<()> { + let api_tx = self + .api_tx + .clone() + .ok_or_else(|| io::Error::other("cannot restore api socket without api sender"))?; + let api_server = api::start_server(api_tx, self.app.event_hub.clone())?; + + let client_path = client_socket_path(); + prepare_socket_path(&client_path)?; + let listener = UnixListener::bind(&client_path)?; + restrict_socket_permissions(&client_path)?; + let client_socket_identity = socket_file_identity(&client_path)?; + listener.set_nonblocking(true)?; + + self.api_server = Some(api_server); + self.client_listener = listener; + self.client_socket_path = client_path; + self.client_socket_identity = client_socket_identity; + Ok(()) + } + + #[cfg(unix)] + fn wait_then_restore_public_sockets_after_failed_handoff(&mut self) -> io::Result<()> { + let timeout = crate::server::handoff::COMMIT_TIMEOUT + Duration::from_secs(2); + wait_for_old_public_sockets_to_close(timeout)?; + self.restore_public_sockets_after_failed_handoff() + } + + #[cfg(unix)] + fn rollback_handoff_before_commit( + &mut self, + socket_path: &Path, + paused_terminal_ids: &[crate::terminal::TerminalId], + ) { + for terminal_id in paused_terminal_ids { + if let Some(runtime) = self.app.terminal_runtimes.get(terminal_id) { + runtime.set_handoff_reader_paused(false); + } + } + self.handoff_in_progress = false; + let _ = std::fs::remove_file(socket_path); + } + + #[cfg(unix)] + fn nudge_handoff_panes_on_first_client_attach(&mut self) { + if !self.pending_handoff_repaint_nudge { + return; + } + self.pending_handoff_repaint_nudge = false; + self.app + .terminal_runtimes + .nudge_child_redraw_after_handoff(); + } + + #[cfg(not(unix))] + fn nudge_handoff_panes_on_first_client_attach(&mut self) {} + fn reload_server_config(&mut self, notify_success: bool) -> crate::config::ConfigReloadReport { let server_keybindings = self.server_keybindings.clone(); apply_keybindings(&mut self.app, &server_keybindings); @@ -706,6 +997,9 @@ impl HeadlessServer { /// Accepts pending client connections from the non-blocking listener. fn accept_client_connections(&mut self) -> io::Result<()> { + if self.handoff_in_progress { + return reject_pending_client_connections(&self.client_listener); + } accept_pending_client_connections( &self.client_listener, &mut self.next_client_id, @@ -1236,6 +1530,28 @@ impl HeadlessServer { } } + fn disconnect_all_clients_for_handoff(&mut self) { + let client_ids = self.clients.keys().copied().collect::>(); + for client_id in client_ids { + self.send_client_graphics_cleanup(client_id); + self.send_to_client( + client_id, + ServerMessage::ServerShutdown { + reason: Some( + "live update in progress; reconnect after handoff completes".to_owned(), + ), + }, + ); + if let Some(client) = self.clients.get_mut(&client_id) { + client.writer = None; + } + let _ = self.remove_client(client_id); + } + self.foreground_client_id = None; + self.sync_foreground_client_state(); + self.resize_shared_runtime_to_effective_size(); + } + fn attach_terminal_client( &mut self, client_id: u64, @@ -1310,6 +1626,10 @@ impl HeadlessServer { /// Handles a server event. Returns true if the event requires a re-render. fn handle_server_event(&mut self, ev: ServerEvent) -> bool { + if self.handoff_in_progress && Self::ignore_client_event_during_handoff(&ev) { + return false; + } + match ev { ServerEvent::ClientConnected { client_id, @@ -1321,6 +1641,19 @@ impl HeadlessServer { writer, render_encoding, } => { + if self.handoff_in_progress { + if let Ok(message) = + Self::frame_server_message(&ServerMessage::ServerShutdown { + reason: Some( + "live update in progress; reconnect after handoff completes" + .to_owned(), + ), + }) + { + let _ = writer.control.send(message); + } + return false; + } info!( client_id, cols, @@ -1351,6 +1684,7 @@ impl HeadlessServer { self.foreground_client_id = Some(client_id); self.sync_foreground_client_state(); self.resize_shared_runtime_to_effective_size(); + self.nudge_handoff_panes_on_first_client_attach(); true } ServerEvent::ClientAttachTerminal { @@ -1370,6 +1704,14 @@ impl HeadlessServer { client_id, source, direction, lines, column, row, modifiers, ), ServerEvent::ClientInput { client_id, data } => { + if self.handoff_in_progress { + debug!( + client_id, + len = data.len(), + "ignored client input during handoff" + ); + return false; + } debug!(client_id, len = data.len(), "client input received"); if let Some(ClientConnection { mode: ClientConnectionMode::TerminalAttach { terminal_id }, @@ -1548,6 +1890,16 @@ impl HeadlessServer { } } + fn ignore_client_event_during_handoff(ev: &ServerEvent) -> bool { + !matches!( + ev, + ServerEvent::ClientConnected { .. } + | ServerEvent::ClientDisconnected { .. } + | ServerEvent::ClientWriterDrained { .. } + | ServerEvent::QuitSignal + ) + } + /// Drains API requests with shutdown awareness. /// /// During shutdown, remaining requests get a `server_unavailable` error. @@ -1583,6 +1935,25 @@ impl HeadlessServer { return false; } + if let api::schema::Method::ServerLiveHandoff(params) = &msg.request.method { + let response = match self.perform_live_handoff(params.clone()) { + Ok(()) => serde_json::to_string(&api::schema::SuccessResponse { + id: msg.request.id, + result: api::schema::ResponseResult::Ok {}, + }), + Err(err) => serde_json::to_string(&api::schema::ErrorResponse { + id: msg.request.id, + error: api::schema::ErrorBody { + code: "handoff_failed".into(), + message: err.to_string(), + }, + }), + } + .unwrap_or_else(|_| "{}".to_string()); + let _ = msg.respond_to.send(response); + return true; + } + let changed = api::request_changes_ui(&msg.request); let changed = self.drain_all_internal_events_with_forwarding() || changed; @@ -2153,7 +2524,9 @@ impl HeadlessServer { /// Removes socket files created by the server. fn cleanup_sockets(&self) -> io::Result<()> { - if let Err(err) = fs::remove_file(&self.client_socket_path) { + if let Err(err) = + remove_socket_file_if_owned(&self.client_socket_path, self.client_socket_identity) + { if err.kind() != io::ErrorKind::NotFound { warn!( path = %self.client_socket_path.display(), @@ -2224,12 +2597,24 @@ fn is_keybinding_config_diagnostic(diagnostic: &str) -> bool { pub fn run_server() -> io::Result<()> { init_logging(); + let args: Vec = std::env::args().collect(); + if args.get(2).map(String::as_str) == Some("--handoff-import") { + let socket_path = args + .get(3) + .map(PathBuf::from) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing handoff socket"))?; + let token = args + .get(4) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing handoff token"))?; + return run_handoff_import_server(&socket_path, token); + } + let loaded_config = config::Config::load(); let (api_tx, api_rx) = tokio::sync::mpsc::unbounded_channel(); let event_hub = api::EventHub::default(); // Start the JSON API socket server. - let _api_server = match api::start_server(api_tx, event_hub.clone()) { + let _api_server = match api::start_server(api_tx.clone(), event_hub.clone()) { Ok(server) => server, Err(err) if err.kind() == io::ErrorKind::AddrInUse => { eprintln!("error: herdr server is already running"); @@ -2263,7 +2648,12 @@ pub fn run_server() -> io::Result<()> { app.local_terminal_notifications = false; // Create the headless server. - let mut server = match HeadlessServer::new(app, &loaded_config.diagnostics) { + let mut server = match HeadlessServer::new( + app, + &loaded_config.diagnostics, + Some(api_tx.clone()), + Some(_api_server), + ) { Ok(server) => server, Err(err) if err.kind() == io::ErrorKind::AddrInUse => { eprintln!("error: herdr server is already running"); @@ -2288,6 +2678,106 @@ pub fn run_server() -> io::Result<()> { result } +#[cfg(unix)] +fn run_handoff_import_server(socket_path: &Path, token: &str) -> io::Result<()> { + let loaded_config = config::Config::load(); + let mut received = crate::server::handoff::receive(socket_path, token)?; + crate::server::handoff::log_import_result(received.manifest.panes.len()); + + let (api_tx, api_rx) = tokio::sync::mpsc::unbounded_channel(); + let event_hub = api::EventHub::default(); + + let mut imports = HashMap::new(); + for (pane, fd) in received.manifest.panes.iter().zip(received.fds) { + imports.insert( + pane.pane_id, + crate::persist::ImportedPaneRuntime { + master_fd: fd, + child_pid: pane.child_pid, + rows: pane.rows, + cols: pane.cols, + cell_width_px: pane.cell_width_px, + cell_height_px: pane.cell_height_px, + initial_history_ansi: pane.initial_history_ansi.clone(), + }, + ); + } + + let rt = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(io::Error::other)?; + + let result = rt.block_on(async { + let mut app = app::App::new_from_handoff( + &loaded_config.config, + config::config_diagnostic_summary(&loaded_config.diagnostics), + api_rx, + event_hub.clone(), + &received.manifest.snapshot, + &mut imports, + )?; + app.state.local_sound_playback = false; + app.local_terminal_notifications = false; + crate::server::handoff::report_restored(&mut received.stream)?; + if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() == Ok("after_restored") { + return Err(io::Error::other( + "test handoff import failure after restored", + )); + } + wait_for_old_public_sockets_to_close(Duration::from_secs(5))?; + + let api_server = api::start_server(api_tx.clone(), event_hub.clone())?; + let mut server = HeadlessServer::new( + app, + &loaded_config.diagnostics, + Some(api_tx.clone()), + Some(api_server), + )?; + crate::server::handoff::report_ready(&mut received.stream)?; + crate::server::handoff::wait_committed(&mut received.stream)?; + server.app.assume_handoff_ownership(); + server.app.unpause_handoff_readers(); + server.pending_handoff_repaint_nudge = true; + if let Err(err) = crate::server::handoff::report_owned(&mut received.stream) { + warn!(err = %err, "failed to report handoff ownership; continuing as owner"); + } + info!("handoff import server started"); + print_ready_message(&api::socket_path(), &client_socket_path()); + server.run().await + }); + + rt.shutdown_timeout(Duration::from_millis(100)); + crate::logging::shutdown("server"); + result +} + +#[cfg(unix)] +fn wait_for_old_public_sockets_to_close(timeout: Duration) -> io::Result<()> { + use std::os::unix::net::UnixStream; + + let deadline = Instant::now() + timeout; + let api_socket = api::socket_path(); + let client_socket = client_socket_path(); + while Instant::now() < deadline { + let api_open = api_socket.exists() && UnixStream::connect(&api_socket).is_ok(); + let client_open = client_socket.exists() && UnixStream::connect(&client_socket).is_ok(); + if !api_open && !client_open { + return Ok(()); + } + std::thread::sleep(Duration::from_millis(50)); + } + Err(io::Error::new( + io::ErrorKind::TimedOut, + "old server sockets did not close before handoff import bind", + )) +} + +#[cfg(not(unix))] +fn run_handoff_import_server(_socket_path: &Path, _token: &str) -> io::Result<()> { + Err(io::Error::other("live handoff is only supported on Unix")) +} + fn print_ready_message(api_socket: &Path, client_socket: &Path) { eprintln!("herdr server running; you can use any herdr CLI command in another terminal."); eprintln!("api socket: {}", api_socket.display()); @@ -2336,6 +2826,8 @@ mod tests { let socket_path = dir.join("client.sock"); let _ = fs::remove_file(&socket_path); let listener = UnixListener::bind(&socket_path).expect("bind test listener"); + let client_socket_identity = + socket_file_identity(&socket_path).expect("test listener socket identity"); listener .set_nonblocking(true) .expect("set listener nonblocking"); @@ -2344,8 +2836,11 @@ mod tests { HeadlessServer { app, + api_tx: None, + api_server: None, client_listener: listener, client_socket_path: socket_path, + client_socket_identity, clients: HashMap::new(), next_client_id: 1, foreground_client_id: None, @@ -2356,6 +2851,8 @@ mod tests { next_activity_stamp: 1, effective_size: (MIN_COLS, MIN_ROWS), shutting_down: false, + handoff_in_progress: false, + pending_handoff_repaint_nudge: false, should_quit: Arc::new(AtomicBool::new(false)), server_event_rx, server_event_tx, diff --git a/src/server/mod.rs b/src/server/mod.rs index b123e111..3386d0d7 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -3,6 +3,8 @@ pub(crate) mod client_accept; pub(crate) mod client_transport; pub(crate) mod clients; pub(crate) mod clipboard_image; +#[cfg(unix)] +pub(crate) mod handoff; pub mod headless; pub(crate) mod keybindings; pub(crate) mod notifications; -- cgit v1.2.3