aboutsummaryrefslogtreecommitdiff
path: root/src/api
diff options
context:
space:
mode:
Diffstat (limited to 'src/api')
-rw-r--r--src/api/client.rs7
-rw-r--r--src/api/mod.rs2
-rw-r--r--src/api/schema.rs20
-rw-r--r--src/api/server.rs128
-rw-r--r--src/api/status.rs8
5 files changed, 138 insertions, 27 deletions
diff --git a/src/api/client.rs b/src/api/client.rs
index 7929f937..7d966965 100644
--- a/src/api/client.rs
+++ b/src/api/client.rs
@@ -111,9 +111,14 @@ impl ApiClient {
method: Method::Ping(PingParams::default()),
})?;
match response.result {
- ResponseResult::Pong { version, protocol } => Ok(crate::api::RuntimeStatus {
+ ResponseResult::Pong {
+ version,
+ protocol,
+ capabilities,
+ } => Ok(crate::api::RuntimeStatus {
version: Some(version),
protocol: Some(protocol),
+ capabilities,
}),
result => Err(ApiClientError::UnexpectedResult(format!("{result:?}"))),
}
diff --git a/src/api/mod.rs b/src/api/mod.rs
index e3ad58c4..1f2ab2bb 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -7,7 +7,7 @@ mod subscriptions;
mod wait;
pub use event_hub::EventHub;
-pub use server::start_server;
+pub use server::{start_server, start_server_with_capabilities, ServerHandle};
pub use status::{read_runtime_status_at, RuntimeStatus};
use std::path::PathBuf;
diff --git a/src/api/schema.rs b/src/api/schema.rs
index 907bc2b6..4929d4a7 100644
--- a/src/api/schema.rs
+++ b/src/api/schema.rs
@@ -14,6 +14,8 @@ pub enum Method {
Ping(PingParams),
#[serde(rename = "server.stop")]
ServerStop(EmptyParams),
+ #[serde(rename = "server.live_handoff")]
+ ServerLiveHandoff(ServerLiveHandoffParams),
#[serde(rename = "server.reload_config")]
ServerReloadConfig(EmptyParams),
#[serde(rename = "workspace.create")]
@@ -307,6 +309,16 @@ pub struct PaneSendInputParams {
pub keys: Vec<String>,
}
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ServerLiveHandoffParams {
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub import_exe: Option<String>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub expected_protocol: Option<u32>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub expected_version: Option<String>,
+}
+
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PaneReadParams {
pub pane_id: String,
@@ -582,11 +594,18 @@ pub struct ErrorBody {
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ServerCapabilities {
+ pub live_handoff: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ResponseResult {
Pong {
version: String,
protocol: u32,
+ #[serde(default)]
+ capabilities: Option<ServerCapabilities>,
},
WorkspaceInfo {
workspace: WorkspaceInfo,
@@ -1225,6 +1244,7 @@ mod tests {
result: ResponseResult::Pong {
version: "0.1.2".into(),
protocol: 6,
+ capabilities: Some(ServerCapabilities { live_handoff: true }),
},
};
diff --git a/src/api/server.rs b/src/api/server.rs
index 513262bc..696e1b65 100644
--- a/src/api/server.rs
+++ b/src/api/server.rs
@@ -1,29 +1,34 @@
-use std::fs;
-use std::io::{BufRead, BufReader, Read, Write};
+use std::io::{self, Read, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
use tracing::{debug, error, info, warn};
+#[cfg(test)]
+use std::fs;
+
use crate::api::schema::{
- ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse,
+ ErrorBody, ErrorResponse, Method, Request, ResponseResult, ServerCapabilities, SuccessResponse,
};
use crate::api::subscriptions::ActiveSubscription;
use crate::api::wait::wait_for_output;
use crate::api::{request_changes_ui, socket_path, ApiRequestMessage, ApiRequestSender, EventHub};
+use crate::ipc::{remove_socket_file_if_owned, socket_file_identity, SocketFileIdentity};
const SOCKET_PERMISSION_MODE: u32 = 0o600;
pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100);
pub(super) const APP_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
const INITIAL_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
const STREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
+const MAX_INITIAL_REQUEST_BYTES: usize = 1024 * 1024;
pub struct ServerHandle {
_thread: std::thread::JoinHandle<()>,
path: PathBuf,
+ identity: SocketFileIdentity,
running: Arc<AtomicBool>,
}
@@ -31,7 +36,7 @@ impl Drop for ServerHandle {
fn drop(&mut self) {
self.running.store(false, Ordering::Relaxed);
- if let Err(err) = fs::remove_file(&self.path) {
+ if let Err(err) = self.remove_socket_file_if_owned() {
if err.kind() != std::io::ErrorKind::NotFound {
warn!(path = %self.path.display(), err = %err, "failed to remove api socket on shutdown");
}
@@ -39,15 +44,34 @@ impl Drop for ServerHandle {
}
}
+impl ServerHandle {
+ pub(crate) fn remove_socket_file_if_owned(&self) -> std::io::Result<()> {
+ remove_socket_file_if_owned(&self.path, self.identity)
+ }
+}
+
pub fn start_server(
api_tx: ApiRequestSender,
event_hub: EventHub,
) -> std::io::Result<ServerHandle> {
+ start_server_with_capabilities(
+ api_tx,
+ event_hub,
+ Some(ServerCapabilities { live_handoff: true }),
+ )
+}
+
+pub fn start_server_with_capabilities(
+ api_tx: ApiRequestSender,
+ event_hub: EventHub,
+ capabilities: Option<ServerCapabilities>,
+) -> std::io::Result<ServerHandle> {
let path = socket_path();
prepare_socket_path(&path)?;
let listener = UnixListener::bind(&path)?;
restrict_socket_permissions(&path)?;
+ let identity = socket_file_identity(&path)?;
info!(path = %path.display(), "api server listening");
let running = Arc::new(AtomicBool::new(true));
@@ -58,11 +82,16 @@ pub fn start_server(
Ok(stream) => {
let api_tx = api_tx.clone();
let event_hub = event_hub.clone();
+ let capabilities = capabilities.clone();
let connection_running = Arc::clone(&listener_running);
std::thread::spawn(move || {
- if let Err(err) =
- handle_connection(stream, &api_tx, &event_hub, &connection_running)
- {
+ if let Err(err) = handle_connection(
+ stream,
+ &api_tx,
+ &event_hub,
+ &connection_running,
+ capabilities,
+ ) {
warn!(err = %err, "api connection failed");
}
});
@@ -79,6 +108,7 @@ pub fn start_server(
Ok(ServerHandle {
_thread: thread,
path,
+ identity,
running,
})
}
@@ -101,20 +131,15 @@ fn handle_connection(
api_tx: &ApiRequestSender,
event_hub: &EventHub,
running: &Arc<AtomicBool>,
+ capabilities: Option<ServerCapabilities>,
) -> std::io::Result<()> {
- stream.set_read_timeout(Some(INITIAL_REQUEST_TIMEOUT))?;
- stream.set_write_timeout(Some(STREAM_WRITE_TIMEOUT))?;
-
- let mut line = String::new();
- {
- let mut reader = BufReader::new(&stream);
- let read = reader.read_line(&mut line)?;
- if read == 0 {
- return Ok(());
- }
+ if let Err(err) = stream.set_write_timeout(Some(STREAM_WRITE_TIMEOUT)) {
+ debug!(err = %err, "api connection write timeout unavailable");
}
- stream.set_read_timeout(None)?;
+ let Some(line) = read_initial_request_line(&mut stream)? else {
+ return Ok(());
+ };
let line = line.trim();
if line.is_empty() {
@@ -199,6 +224,7 @@ fn handle_connection(
method: method_body,
},
api_tx,
+ capabilities,
);
let result = write_text_line_allow_disconnect(&mut stream, &response);
match &result {
@@ -217,13 +243,18 @@ fn handle_connection(
}
}
-fn handle_request(request: Request, api_tx: &ApiRequestSender) -> String {
+fn handle_request(
+ request: Request,
+ api_tx: &ApiRequestSender,
+ capabilities: Option<ServerCapabilities>,
+) -> String {
match request.method {
Method::Ping(_) => serde_json::to_string(&SuccessResponse {
id: request.id,
result: ResponseResult::Pong {
version: env!("CARGO_PKG_VERSION").into(),
protocol: crate::protocol::PROTOCOL_VERSION,
+ capabilities,
},
})
.unwrap_or_else(|_| {
@@ -238,6 +269,7 @@ fn api_method_name(method: &Method) -> &'static str {
match method {
Method::Ping(_) => "ping",
Method::ServerStop(_) => "server.stop",
+ Method::ServerLiveHandoff(_) => "server.live_handoff",
Method::ServerReloadConfig(_) => "server.reload_config",
Method::WorkspaceCreate(_) => "workspace.create",
Method::WorkspaceList(_) => "workspace.list",
@@ -298,6 +330,52 @@ fn api_response_outcome(response: &str) -> &'static str {
}
}
+fn read_initial_request_line(stream: &mut UnixStream) -> std::io::Result<Option<String>> {
+ stream.set_nonblocking(true)?;
+ let deadline = Instant::now() + INITIAL_REQUEST_TIMEOUT;
+ let mut bytes = Vec::new();
+ let mut byte = [0u8; 1];
+
+ loop {
+ match stream.read(&mut byte) {
+ Ok(0) => {
+ stream.set_nonblocking(false)?;
+ return Ok(None);
+ }
+ Ok(_) => {
+ bytes.push(byte[0]);
+ if byte[0] == b'\n' {
+ stream.set_nonblocking(false)?;
+ return String::from_utf8(bytes)
+ .map(Some)
+ .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err));
+ }
+ if bytes.len() > MAX_INITIAL_REQUEST_BYTES {
+ stream.set_nonblocking(false)?;
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "api request line is too large",
+ ));
+ }
+ }
+ Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
+ if Instant::now() >= deadline {
+ stream.set_nonblocking(false)?;
+ return Err(io::Error::new(
+ io::ErrorKind::TimedOut,
+ "timed out reading api request",
+ ));
+ }
+ std::thread::sleep(CONNECTION_POLL_INTERVAL);
+ }
+ Err(err) => {
+ stream.set_nonblocking(false)?;
+ return Err(err);
+ }
+ }
+ }
+}
+
fn stream_subscriptions(
mut stream: UnixStream,
request_id: String,
@@ -496,6 +574,7 @@ fn error_response_json(id: String, code: &str, message: String) -> String {
#[cfg(test)]
mod tests {
use super::*;
+ use std::io::{BufRead, BufReader};
use std::os::unix::fs::PermissionsExt;
use std::sync::{Mutex, OnceLock};
use tokio::sync::mpsc;
@@ -610,6 +689,7 @@ mod tests {
method: Method::Ping(crate::api::schema::PingParams::default()),
},
&tx,
+ Some(ServerCapabilities { live_handoff: true }),
);
let parsed: SuccessResponse = serde_json::from_str(&response).unwrap();
@@ -626,7 +706,7 @@ mod tests {
};
let request_for_thread = request.clone();
- let thread = std::thread::spawn(move || handle_request(request_for_thread, &tx));
+ let thread = std::thread::spawn(move || handle_request(request_for_thread, &tx, None));
let msg = rx.blocking_recv().unwrap();
assert_eq!(msg.request.id, "req_2");
@@ -692,7 +772,7 @@ mod tests {
let event_hub = EventHub::default();
let (done_tx, done_rx) = std::sync::mpsc::channel();
let server_thread = std::thread::spawn(move || {
- let result = handle_connection(server, &api_tx, &event_hub, &server_running);
+ let result = handle_connection(server, &api_tx, &event_hub, &server_running, None);
done_tx.send(result).unwrap();
});
@@ -724,7 +804,7 @@ mod tests {
let event_hub = EventHub::default();
let (done_tx, done_rx) = std::sync::mpsc::channel();
let server_thread = std::thread::spawn(move || {
- let result = handle_connection(server, &api_tx, &event_hub, &server_running);
+ let result = handle_connection(server, &api_tx, &event_hub, &server_running, None);
done_tx.send(result).unwrap();
});
@@ -756,7 +836,7 @@ mod tests {
let event_hub = EventHub::default();
let (done_tx, done_rx) = std::sync::mpsc::channel();
let server_thread = std::thread::spawn(move || {
- let result = handle_connection(server, &api_tx, &event_hub, &server_running);
+ let result = handle_connection(server, &api_tx, &event_hub, &server_running, None);
done_tx.send(result).unwrap();
});
diff --git a/src/api/status.rs b/src/api/status.rs
index f58e69c0..209b93df 100644
--- a/src/api/status.rs
+++ b/src/api/status.rs
@@ -8,6 +8,7 @@ use crate::api::schema::{Method, Request, ResponseResult};
pub struct RuntimeStatus {
pub version: Option<String>,
pub protocol: Option<u32>,
+ pub capabilities: Option<crate::api::schema::ServerCapabilities>,
}
pub fn read_runtime_status_at(
@@ -43,9 +44,14 @@ pub fn read_runtime_status_at(
Err(err) => return Err(io::Error::other(err)),
};
match response.result {
- ResponseResult::Pong { version, protocol } => Ok(Some(RuntimeStatus {
+ ResponseResult::Pong {
+ version,
+ protocol,
+ capabilities,
+ } => Ok(Some(RuntimeStatus {
version: Some(version),
protocol: Some(protocol),
+ capabilities,
})),
result => Err(io::Error::other(format!(
"server status request returned unexpected result: {result:?}"
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-06 22:35:05 +0000