From 6bc3804168af7d694a6c3461a9f5adc7f590e3e4 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Fri, 16 Oct 2026 07:07:17 +0000 Subject: [PATCH] spaced-host-client: move the client's methods into modules by guest area (pure move) lib.rs holds the whole typed client: its types, the call macro and constructors, and 40 methods across four guest services. Keep the types, the macro and the constructors in lib.rs and move the methods into impl blocks in five child modules: meta, filesystem, file_watch, process and networking, together with the helpers only those methods use. Generated by scripts/split_rust_module.py (#562) from the spec in the pull request. Items move verbatim. The only edits keep each item's reach: 5 private items, fields or methods become pub(super). The parent re-exports the moved items, so no caller changes. --- .../libs/spaced/host/client/src/file_watch.rs | 224 +++++ .../libs/spaced/host/client/src/filesystem.rs | 335 ++++++ crates/libs/spaced/host/client/src/lib.rs | 950 +----------------- crates/libs/spaced/host/client/src/meta.rs | 113 +++ .../libs/spaced/host/client/src/networking.rs | 47 + crates/libs/spaced/host/client/src/process.rs | 248 +++++ 6 files changed, 978 insertions(+), 939 deletions(-) create mode 100644 crates/libs/spaced/host/client/src/file_watch.rs create mode 100644 crates/libs/spaced/host/client/src/filesystem.rs create mode 100644 crates/libs/spaced/host/client/src/meta.rs create mode 100644 crates/libs/spaced/host/client/src/networking.rs create mode 100644 crates/libs/spaced/host/client/src/process.rs diff --git a/crates/libs/spaced/host/client/src/file_watch.rs b/crates/libs/spaced/host/client/src/file_watch.rs new file mode 100644 index 000000000..92862d246 --- /dev/null +++ b/crates/libs/spaced/host/client/src/file_watch.rs @@ -0,0 +1,224 @@ +//! File watching and change tracking. + +use super::*; + +impl SpacedHostClient { + pub async fn watch_stream( + &self, + path: String, + recursive: bool, + ) -> Result, tonic::Status> { + self.watch_stream_with_options( + path, + WatchOptions { + recursive, + include_metadata_events: true, + ..Default::default() + }, + ) + .await + } + + pub async fn watch_stream_with_options( + &self, + path: String, + options: WatchOptions, + ) -> Result, tonic::Status> { + let root = WatchRoot::new(GuestPath::parse(path).map_err(invalid_argument_status)?); + let include_globs = options.include_globs; + let exclude_globs = options.exclude_globs; + let recursive = options.recursive; + let include_metadata_events = options.include_metadata_events; + let use_default_excludes = options.use_default_excludes; + let user = options.user; + pplx_space_spaced_common::retry_status( + "spaced host", + "watch", + self.config.command_retry_policy, + || { + let mut client = self.filesystem.clone(); + let path = root.guest_path().expose().to_string(); + let include_globs = include_globs.clone(); + let exclude_globs = exclude_globs.clone(); + let user = user.clone(); + async move { + client + .watch(WatchRequest { + path, + recursive, + include_metadata_events, + include_globs: proto_globs(&include_globs), + exclude_globs: proto_globs(&exclude_globs), + use_default_excludes, + user: user.map(|user| user.as_str().to_string()), + }) + .await + .map(Response::into_inner) + } + }, + ) + .await + } + + pub async fn create_watcher_with_options( + &self, + root: String, + options: WatchOptions, + ) -> Result { + let root = WatchRoot::new(GuestPath::parse(root).map_err(invalid_argument_status)?); + let include_globs = options.include_globs; + let exclude_globs = options.exclude_globs; + let recursive = options.recursive; + let include_metadata_events = options.include_metadata_events; + let use_default_excludes = options.use_default_excludes; + let user = options.user; + self.call_filesystem_once("create_watcher", |mut client| { + let root = root.guest_path().expose().to_string(); + let include_globs = include_globs.clone(); + let exclude_globs = exclude_globs.clone(); + let user = user.clone(); + async move { + client + .create_watcher(CreateWatcherRequest { + root, + recursive, + include_metadata_events, + include_globs: proto_globs(&include_globs), + exclude_globs: proto_globs(&exclude_globs), + use_default_excludes, + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn normalize_watch_root_as_user( + &self, + path: String, + user: Option, + ) -> Result { + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + self.call_filesystem("normalize_watch_root", |mut client| { + let path = path.clone(); + let user = user.clone(); + async move { + client + .normalize_watch_root(NormalizeWatchRootRequest { + path: path.expose().to_string(), + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn get_watcher_events( + &self, + watcher_id: String, + ) -> Result { + self.call_filesystem_once("get_watcher_events", |mut client| { + let watcher_id = watcher_id.clone(); + async move { + client + .get_watcher_events(GetWatcherEventsRequest { watcher_id }) + .await + } + }) + .await + } + + pub async fn remove_watcher( + &self, + watcher_id: String, + ) -> Result { + self.call_filesystem_once("remove_watcher", |mut client| { + let watcher_id = watcher_id.clone(); + async move { + client + .remove_watcher(RemoveWatcherRequest { watcher_id }) + .await + } + }) + .await + } + + pub async fn enable_file_changes_with_options_as_user( + &self, + roots: Vec, + include_globs: Vec, + exclude_globs: Vec, + use_default_excludes: Option, + user: Option, + ) -> Result { + let roots = roots + .into_iter() + .map(GuestPath::parse) + .collect::, _>>() + .map_err(invalid_argument_status)?; + self.call_filesystem_once("enable_file_changes", |mut client| { + let roots = roots + .iter() + .map(|root| root.expose().to_string()) + .collect::>(); + let include_globs = include_globs.clone(); + let exclude_globs = exclude_globs.clone(); + let user = user.clone(); + async move { + client + .enable_file_changes(EnableFileChangesRequest { + roots, + include_globs: proto_globs(&include_globs), + exclude_globs: proto_globs(&exclude_globs), + use_default_excludes, + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn disable_file_changes_as_user( + &self, + user: Option, + ) -> Result { + self.call_filesystem_once("disable_file_changes", |mut client| { + let user = user.clone(); + async move { + client + .disable_file_changes(DisableFileChangesRequest { + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn get_modified_files_as_user( + &self, + window: Option, + since: Option, + cursor: Option, + user: Option, + ) -> Result { + self.call_filesystem("get_modified_files", |mut client| { + let cursor = cursor.clone(); + let user = user.clone(); + async move { + client + .get_modified_files(GetModifiedFilesRequest { + window, + since, + cursor, + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } +} diff --git a/crates/libs/spaced/host/client/src/filesystem.rs b/crates/libs/spaced/host/client/src/filesystem.rs new file mode 100644 index 000000000..b57e8ed7c --- /dev/null +++ b/crates/libs/spaced/host/client/src/filesystem.rs @@ -0,0 +1,335 @@ +//! Filesystem calls: stat, list, read, write, mkdir, remove, rename and batched writes. + +use super::*; + +pub(super) fn idempotency_key_or_generated( + key: Option, + required: bool, +) -> Result, tonic::Status> { + match key { + Some(key) => IdempotencyKey::parse(key) + .map(Some) + .map_err(invalid_argument_status), + None if required => IdempotencyKey::parse(uuid::Uuid::now_v7().to_string()) + .map(Some) + .map_err(invalid_argument_status), + None => Ok(None), + } +} + +impl SpacedHostClient { + pub async fn stat(&self, path: String) -> Result { + self.stat_as_user(path, None).await + } + + pub async fn stat_as_user( + &self, + path: String, + user: Option, + ) -> Result { + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + self.call_filesystem("stat", |mut client| { + let path = path.clone(); + let user = user.clone(); + async move { + client + .stat(StatRequest { + path: path.expose().to_string(), + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn list_stream_with_depth_as_user( + &self, + path: String, + depth: Option, + user: Option, + ) -> Result, tonic::Status> { + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + if depth == Some(0) { + return Err(tonic::Status::invalid_argument( + "depth must be greater than 0", + )); + } + pplx_space_spaced_common::retry_status( + "spaced host", + "list", + self.config.command_retry_policy, + || { + let mut client = self.filesystem.clone(); + let path = path.clone(); + let user = user.clone(); + async move { + client + .list(ListRequest { + path: path.expose().to_string(), + depth, + user: user.map(|user| user.as_str().to_string()), + }) + .await + .map(Response::into_inner) + } + }, + ) + .await + } + + pub async fn read_stream_as_user( + &self, + path: String, + offset: u64, + length: Option, + user: Option, + ) -> Result, tonic::Status> { + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + pplx_space_spaced_common::retry_status( + "spaced host", + "read", + self.config.command_retry_policy, + || { + let mut client = self.filesystem.clone(); + let path = path.clone(); + let user = user.clone(); + async move { + client + .read(proto::ReadRequest { + path: path.expose().to_string(), + offset, + length, + user: user.map(|user| user.as_str().to_string()), + }) + .await + .map(Response::into_inner) + } + }, + ) + .await + } + + pub async fn write( + &self, + path: String, + data: Vec, + append: bool, + ) -> Result { + self.write_with_idempotency(path, data, append, None, None) + .await + } + + pub async fn write_with_idempotency( + &self, + path: String, + data: Vec, + append: bool, + idempotency_key: Option, + user: Option, + ) -> Result { + const CHUNK_SIZE: usize = 64 * 1024; + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + let idempotency_key = idempotency_key_or_generated(idempotency_key, append)?; + let data = std::sync::Arc::new(data); + + pplx_space_spaced_common::retry_status( + "spaced host", + "write", + self.config.command_retry_policy, + || { + let mut client = self.filesystem.clone(); + let path = path.clone(); + let idempotency_key = idempotency_key.clone(); + let user = user.clone(); + let data = data.clone(); + async move { + let open = WriteRequest { + message: Some(proto::write_request::Message::Open(proto::WriteOpen { + path: path.expose().to_string(), + append, + idempotency_key: proto_idempotency_key(idempotency_key), + user: user.map(|user| user.as_str().to_string()), + })), + }; + // Retries share the supplied payload; only the next frame + // is copied, not the complete upload on every attempt. + let chunks = + futures_util::stream::unfold((data, 0), |(data, offset)| async move { + let chunk = data.get(offset..)?.chunks(CHUNK_SIZE).next()?.to_vec(); + let next = offset + chunk.len(); + Some(( + WriteRequest { + message: Some(proto::write_request::Message::Data(chunk)), + }, + (data, next), + )) + }); + let messages = futures_util::stream::once(async move { open }).chain(chunks); + client.write(messages).await.map(Response::into_inner) + } + }, + ) + .await + } + + pub async fn write_stream_with_idempotency( + &self, + path: String, + append: bool, + idempotency_key: Option, + user: Option, + data: S, + ) -> Result + where + S: futures_util::Stream> + Send + 'static, + { + use futures_util::StreamExt; + + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + let idempotency_key = idempotency_key_or_generated(idempotency_key, append)?; + let open = WriteRequest { + message: Some(proto::write_request::Message::Open(proto::WriteOpen { + path: path.expose().to_string(), + append, + idempotency_key: proto_idempotency_key(idempotency_key), + user: user.map(|user| user.as_str().to_string()), + })), + }; + let messages = + futures_util::stream::once(async move { open }).chain(data.map(|chunk| WriteRequest { + message: Some(proto::write_request::Message::Data(chunk)), + })); + self.filesystem + .clone() + .write(messages) + .await + .map(Response::into_inner) + } + + pub async fn mkdir( + &self, + path: String, + parents: bool, + exist_ok: bool, + user: Option, + ) -> Result { + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + self.call_filesystem_once("mkdir", |mut client| { + let path = path.clone(); + let user = user.clone(); + async move { + client + .mkdir(MkdirRequest { + path: path.expose().to_string(), + parents, + exist_ok, + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn remove( + &self, + path: String, + recursive: bool, + missing_ok: bool, + ) -> Result { + self.remove_as_user(path, recursive, missing_ok, None).await + } + + pub async fn remove_as_user( + &self, + path: String, + recursive: bool, + missing_ok: bool, + user: Option, + ) -> Result { + let path = GuestPath::parse(path).map_err(invalid_argument_status)?; + self.call_filesystem_once("remove", |mut client| { + let path = path.clone(); + let user = user.clone(); + async move { + client + .remove(RemoveRequest { + path: path.expose().to_string(), + recursive, + missing_ok, + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn rename_as_user( + &self, + old_path: String, + new_path: String, + idempotency_key: Option, + user: Option, + ) -> Result { + let old_path = GuestPath::parse(old_path).map_err(invalid_argument_status)?; + let new_path = GuestPath::parse(new_path).map_err(invalid_argument_status)?; + let idempotency_key = idempotency_key_or_generated(idempotency_key, true)?; + self.call_filesystem("rename", |mut client| { + let old_path = old_path.clone(); + let new_path = new_path.clone(); + let idempotency_key = idempotency_key.clone(); + let user = user.clone(); + async move { + client + .rename(RenameRequest { + old_path: old_path.expose().to_string(), + new_path: new_path.expose().to_string(), + idempotency_key: proto_idempotency_key(idempotency_key), + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } + + pub async fn write_batch_as_user( + &self, + entries: Vec, + idempotency_key: Option, + user: Option, + ) -> Result { + let entries = entries + .into_iter() + .map(|entry| { + let path = GuestPath::parse(entry.path).map_err(invalid_argument_status)?; + Ok((path, entry.data, entry.append)) + }) + .collect::, tonic::Status>>()?; + let idempotency_key = idempotency_key_or_generated(idempotency_key, true)?; + self.call_filesystem("write_batch", |mut client| { + let entries = entries.clone(); + let idempotency_key = idempotency_key.clone(); + let user = user.clone(); + async move { + client + .write_batch(WriteBatchRequest { + entries: entries + .into_iter() + .map(|(path, data, append)| WriteBatchEntry { + path: path.expose().to_string(), + data, + append, + }) + .collect(), + idempotency_key: proto_idempotency_key(idempotency_key), + user: user.map(|user| user.as_str().to_string()), + }) + .await + } + }) + .await + } +} diff --git a/crates/libs/spaced/host/client/src/lib.rs b/crates/libs/spaced/host/client/src/lib.rs index c525a8e09..03b310529 100644 --- a/crates/libs/spaced/host/client/src/lib.rs +++ b/crates/libs/spaced/host/client/src/lib.rs @@ -50,43 +50,12 @@ pub use pplx_space_spaced_common::{ DEFAULT_COMMAND_RETRY_POLICY, DEFAULT_CONNECT_RETRY_POLICY, RetryPolicy, }; -const PROCESS_LIST_MAX_RECORDS: usize = 128; -const PROCESS_LIST_MAX_ENCODED_BYTES: usize = 4 * 1024 * 1024; -const PROCESS_LIST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); - pub fn default_host_uds_path() -> std::path::PathBuf { pplx_space_common_core::host::linux::paths::default_host_uds_path( pplx_space_common_core::host::linux::paths::SPACED_HOST_SOCKET_FILE, ) } -fn idempotency_key_or_generated( - key: Option, - required: bool, -) -> Result, tonic::Status> { - match key { - Some(key) => IdempotencyKey::parse(key) - .map(Some) - .map_err(invalid_argument_status), - None if required => IdempotencyKey::parse(uuid::Uuid::now_v7().to_string()) - .map(Some) - .map_err(invalid_argument_status), - None => Ok(None), - } -} - -fn connect_process_request( - process_id: &ProcessId, - timeout: Option, - output_cursor: Option, -) -> ConnectProcessRequest { - ConnectProcessRequest { - process_id: process_id.as_str().to_string(), - timeout: timeout.map(proto_duration), - output_cursor, - } -} - #[derive(Clone, Debug)] pub struct SpacedHostClient { meta: RawSpacedHostMetaClient, @@ -176,6 +145,17 @@ macro_rules! client_call { }; } +mod file_watch; +mod filesystem; +mod meta; +mod networking; +mod process; + +#[allow(unused_imports)] +use filesystem::*; +#[allow(unused_imports)] +use process::*; + impl SpacedHostClient { client_call!(call_process, process, RawSpacedHostProcessClient, false); client_call!(call_meta, meta, RawSpacedHostMetaClient); @@ -234,914 +214,6 @@ impl SpacedHostClient { config, } } - - pub async fn ping(&self) -> Result { - self.call_meta("ping", |mut client| async move { - client.ping(PingRequest {}).await - }) - .await - } - - pub async fn is_connected(&self) -> Result { - self.call_meta("is_connected", |mut client| async move { - client.is_connected(IsConnectedRequest {}).await - }) - .await - } - - pub async fn ping_guest(&self) -> Result { - self.call_meta("ping_guest", |mut client| async move { - client.ping_guest(PingGuestRequest {}).await - }) - .await - } - - pub async fn get_status(&self) -> Result { - self.call_meta("get_status", |mut client| async move { - client.get_status(GetStatusRequest {}).await - }) - .await - } - - pub async fn wait_ready( - &self, - timeout: std::time::Duration, - ) -> Result { - let timeout = ReadyTimeout::new(timeout).map_err(invalid_argument_status)?; - self.call_meta("wait_ready", |mut client| async move { - client - .wait_ready(WaitReadyRequest { - timeout: Some(proto_duration(timeout)), - }) - .await - }) - .await - } - - pub async fn configure_process_env( - &self, - env: std::collections::HashMap, - ) -> Result<(), tonic::Status> { - let env = parse_env(env); - self.call_meta_once("configure_process_env", |mut client| { - let env = proto_env(&env); - async move { - client - .configure_process_env(ConfigureProcessEnvRequest { env }) - .await - } - }) - .await - .map(|_: proto::ConfigureProcessEnvResponse| ()) - } - - pub async fn notify_resume( - &self, - reason: ResumeReason, - ) -> Result { - let now = pplx_space_spaced_common::system_time_to_proto(std::time::SystemTime::now())?; - self.call_meta_once("notify_resume", |mut client| async move { - client - .notify_resume(NotifyResumeRequest { - reason: reason.into(), - host_time: Some(now), - }) - .await - }) - .await - } - - /// Prepare for a VM memory snapshot, including the final guest filesystem flush. - /// - /// Call this instead of `flush_guest_filesystems` for suspend or full - /// checkpoint paths; do not call both for one snapshot. - pub async fn prepare_suspend(&self, timeout: std::time::Duration) -> Result<(), tonic::Status> { - self.call_meta_once("prepare_suspend", |mut client| async move { - let mut request = tonic::Request::new(PrepareSuspendRequest {}); - request.set_timeout(timeout.min(std::time::Duration::from_secs(180))); - client.prepare_suspend(request).await - }) - .await - .map(|_: proto::PrepareSuspendResponse| ()) - } - - /// Flush guest filesystems without preparing bridge connections for a VM snapshot. - /// - /// Use this for disk-only snapshots of a still-running VM. Memory-bearing - /// snapshots must call `prepare_suspend` instead. - pub async fn flush_guest_filesystems( - &self, - timeout: std::time::Duration, - ) -> Result<(), tonic::Status> { - self.call_meta_once("flush_guest_filesystems", |mut client| async move { - let mut request = tonic::Request::new(FlushGuestFilesystemsRequest {}); - request.set_timeout(timeout.min(std::time::Duration::from_secs(180))); - client.flush_guest_filesystems(request).await - }) - .await - .map(|_: proto::FlushGuestFilesystemsResponse| ()) - } - - pub async fn stat(&self, path: String) -> Result { - self.stat_as_user(path, None).await - } - - pub async fn stat_as_user( - &self, - path: String, - user: Option, - ) -> Result { - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - self.call_filesystem("stat", |mut client| { - let path = path.clone(); - let user = user.clone(); - async move { - client - .stat(StatRequest { - path: path.expose().to_string(), - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn list_stream_with_depth_as_user( - &self, - path: String, - depth: Option, - user: Option, - ) -> Result, tonic::Status> { - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - if depth == Some(0) { - return Err(tonic::Status::invalid_argument( - "depth must be greater than 0", - )); - } - pplx_space_spaced_common::retry_status( - "spaced host", - "list", - self.config.command_retry_policy, - || { - let mut client = self.filesystem.clone(); - let path = path.clone(); - let user = user.clone(); - async move { - client - .list(ListRequest { - path: path.expose().to_string(), - depth, - user: user.map(|user| user.as_str().to_string()), - }) - .await - .map(Response::into_inner) - } - }, - ) - .await - } - - pub async fn read_stream_as_user( - &self, - path: String, - offset: u64, - length: Option, - user: Option, - ) -> Result, tonic::Status> { - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - pplx_space_spaced_common::retry_status( - "spaced host", - "read", - self.config.command_retry_policy, - || { - let mut client = self.filesystem.clone(); - let path = path.clone(); - let user = user.clone(); - async move { - client - .read(proto::ReadRequest { - path: path.expose().to_string(), - offset, - length, - user: user.map(|user| user.as_str().to_string()), - }) - .await - .map(Response::into_inner) - } - }, - ) - .await - } - - pub async fn write( - &self, - path: String, - data: Vec, - append: bool, - ) -> Result { - self.write_with_idempotency(path, data, append, None, None) - .await - } - - pub async fn write_with_idempotency( - &self, - path: String, - data: Vec, - append: bool, - idempotency_key: Option, - user: Option, - ) -> Result { - const CHUNK_SIZE: usize = 64 * 1024; - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - let idempotency_key = idempotency_key_or_generated(idempotency_key, append)?; - let data = std::sync::Arc::new(data); - - pplx_space_spaced_common::retry_status( - "spaced host", - "write", - self.config.command_retry_policy, - || { - let mut client = self.filesystem.clone(); - let path = path.clone(); - let idempotency_key = idempotency_key.clone(); - let user = user.clone(); - let data = data.clone(); - async move { - let open = WriteRequest { - message: Some(proto::write_request::Message::Open(proto::WriteOpen { - path: path.expose().to_string(), - append, - idempotency_key: proto_idempotency_key(idempotency_key), - user: user.map(|user| user.as_str().to_string()), - })), - }; - // Retries share the supplied payload; only the next frame - // is copied, not the complete upload on every attempt. - let chunks = - futures_util::stream::unfold((data, 0), |(data, offset)| async move { - let chunk = data.get(offset..)?.chunks(CHUNK_SIZE).next()?.to_vec(); - let next = offset + chunk.len(); - Some(( - WriteRequest { - message: Some(proto::write_request::Message::Data(chunk)), - }, - (data, next), - )) - }); - let messages = futures_util::stream::once(async move { open }).chain(chunks); - client.write(messages).await.map(Response::into_inner) - } - }, - ) - .await - } - - pub async fn write_stream_with_idempotency( - &self, - path: String, - append: bool, - idempotency_key: Option, - user: Option, - data: S, - ) -> Result - where - S: futures_util::Stream> + Send + 'static, - { - use futures_util::StreamExt; - - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - let idempotency_key = idempotency_key_or_generated(idempotency_key, append)?; - let open = WriteRequest { - message: Some(proto::write_request::Message::Open(proto::WriteOpen { - path: path.expose().to_string(), - append, - idempotency_key: proto_idempotency_key(idempotency_key), - user: user.map(|user| user.as_str().to_string()), - })), - }; - let messages = - futures_util::stream::once(async move { open }).chain(data.map(|chunk| WriteRequest { - message: Some(proto::write_request::Message::Data(chunk)), - })); - self.filesystem - .clone() - .write(messages) - .await - .map(Response::into_inner) - } - - pub async fn mkdir( - &self, - path: String, - parents: bool, - exist_ok: bool, - user: Option, - ) -> Result { - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - self.call_filesystem_once("mkdir", |mut client| { - let path = path.clone(); - let user = user.clone(); - async move { - client - .mkdir(MkdirRequest { - path: path.expose().to_string(), - parents, - exist_ok, - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn remove( - &self, - path: String, - recursive: bool, - missing_ok: bool, - ) -> Result { - self.remove_as_user(path, recursive, missing_ok, None).await - } - - pub async fn remove_as_user( - &self, - path: String, - recursive: bool, - missing_ok: bool, - user: Option, - ) -> Result { - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - self.call_filesystem_once("remove", |mut client| { - let path = path.clone(); - let user = user.clone(); - async move { - client - .remove(RemoveRequest { - path: path.expose().to_string(), - recursive, - missing_ok, - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn rename_as_user( - &self, - old_path: String, - new_path: String, - idempotency_key: Option, - user: Option, - ) -> Result { - let old_path = GuestPath::parse(old_path).map_err(invalid_argument_status)?; - let new_path = GuestPath::parse(new_path).map_err(invalid_argument_status)?; - let idempotency_key = idempotency_key_or_generated(idempotency_key, true)?; - self.call_filesystem("rename", |mut client| { - let old_path = old_path.clone(); - let new_path = new_path.clone(); - let idempotency_key = idempotency_key.clone(); - let user = user.clone(); - async move { - client - .rename(RenameRequest { - old_path: old_path.expose().to_string(), - new_path: new_path.expose().to_string(), - idempotency_key: proto_idempotency_key(idempotency_key), - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn write_batch_as_user( - &self, - entries: Vec, - idempotency_key: Option, - user: Option, - ) -> Result { - let entries = entries - .into_iter() - .map(|entry| { - let path = GuestPath::parse(entry.path).map_err(invalid_argument_status)?; - Ok((path, entry.data, entry.append)) - }) - .collect::, tonic::Status>>()?; - let idempotency_key = idempotency_key_or_generated(idempotency_key, true)?; - self.call_filesystem("write_batch", |mut client| { - let entries = entries.clone(); - let idempotency_key = idempotency_key.clone(); - let user = user.clone(); - async move { - client - .write_batch(WriteBatchRequest { - entries: entries - .into_iter() - .map(|(path, data, append)| WriteBatchEntry { - path: path.expose().to_string(), - data, - append, - }) - .collect(), - idempotency_key: proto_idempotency_key(idempotency_key), - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn watch_stream( - &self, - path: String, - recursive: bool, - ) -> Result, tonic::Status> { - self.watch_stream_with_options( - path, - WatchOptions { - recursive, - include_metadata_events: true, - ..Default::default() - }, - ) - .await - } - - pub async fn watch_stream_with_options( - &self, - path: String, - options: WatchOptions, - ) -> Result, tonic::Status> { - let root = WatchRoot::new(GuestPath::parse(path).map_err(invalid_argument_status)?); - let include_globs = options.include_globs; - let exclude_globs = options.exclude_globs; - let recursive = options.recursive; - let include_metadata_events = options.include_metadata_events; - let use_default_excludes = options.use_default_excludes; - let user = options.user; - pplx_space_spaced_common::retry_status( - "spaced host", - "watch", - self.config.command_retry_policy, - || { - let mut client = self.filesystem.clone(); - let path = root.guest_path().expose().to_string(); - let include_globs = include_globs.clone(); - let exclude_globs = exclude_globs.clone(); - let user = user.clone(); - async move { - client - .watch(WatchRequest { - path, - recursive, - include_metadata_events, - include_globs: proto_globs(&include_globs), - exclude_globs: proto_globs(&exclude_globs), - use_default_excludes, - user: user.map(|user| user.as_str().to_string()), - }) - .await - .map(Response::into_inner) - } - }, - ) - .await - } - - pub async fn create_watcher_with_options( - &self, - root: String, - options: WatchOptions, - ) -> Result { - let root = WatchRoot::new(GuestPath::parse(root).map_err(invalid_argument_status)?); - let include_globs = options.include_globs; - let exclude_globs = options.exclude_globs; - let recursive = options.recursive; - let include_metadata_events = options.include_metadata_events; - let use_default_excludes = options.use_default_excludes; - let user = options.user; - self.call_filesystem_once("create_watcher", |mut client| { - let root = root.guest_path().expose().to_string(); - let include_globs = include_globs.clone(); - let exclude_globs = exclude_globs.clone(); - let user = user.clone(); - async move { - client - .create_watcher(CreateWatcherRequest { - root, - recursive, - include_metadata_events, - include_globs: proto_globs(&include_globs), - exclude_globs: proto_globs(&exclude_globs), - use_default_excludes, - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn normalize_watch_root_as_user( - &self, - path: String, - user: Option, - ) -> Result { - let path = GuestPath::parse(path).map_err(invalid_argument_status)?; - self.call_filesystem("normalize_watch_root", |mut client| { - let path = path.clone(); - let user = user.clone(); - async move { - client - .normalize_watch_root(NormalizeWatchRootRequest { - path: path.expose().to_string(), - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn get_watcher_events( - &self, - watcher_id: String, - ) -> Result { - self.call_filesystem_once("get_watcher_events", |mut client| { - let watcher_id = watcher_id.clone(); - async move { - client - .get_watcher_events(GetWatcherEventsRequest { watcher_id }) - .await - } - }) - .await - } - - pub async fn remove_watcher( - &self, - watcher_id: String, - ) -> Result { - self.call_filesystem_once("remove_watcher", |mut client| { - let watcher_id = watcher_id.clone(); - async move { - client - .remove_watcher(RemoveWatcherRequest { watcher_id }) - .await - } - }) - .await - } - - pub async fn enable_file_changes_with_options_as_user( - &self, - roots: Vec, - include_globs: Vec, - exclude_globs: Vec, - use_default_excludes: Option, - user: Option, - ) -> Result { - let roots = roots - .into_iter() - .map(GuestPath::parse) - .collect::, _>>() - .map_err(invalid_argument_status)?; - self.call_filesystem_once("enable_file_changes", |mut client| { - let roots = roots - .iter() - .map(|root| root.expose().to_string()) - .collect::>(); - let include_globs = include_globs.clone(); - let exclude_globs = exclude_globs.clone(); - let user = user.clone(); - async move { - client - .enable_file_changes(EnableFileChangesRequest { - roots, - include_globs: proto_globs(&include_globs), - exclude_globs: proto_globs(&exclude_globs), - use_default_excludes, - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn disable_file_changes_as_user( - &self, - user: Option, - ) -> Result { - self.call_filesystem_once("disable_file_changes", |mut client| { - let user = user.clone(); - async move { - client - .disable_file_changes(DisableFileChangesRequest { - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn get_modified_files_as_user( - &self, - window: Option, - since: Option, - cursor: Option, - user: Option, - ) -> Result { - self.call_filesystem("get_modified_files", |mut client| { - let cursor = cursor.clone(); - let user = user.clone(); - async move { - client - .get_modified_files(GetModifiedFilesRequest { - window, - since, - cursor, - user: user.map(|user| user.as_str().to_string()), - }) - .await - } - }) - .await - } - - pub async fn start_process_stream_with_options( - &self, - options: StartProcessOptions, - ) -> Result, tonic::Status> { - let tag = options.tag.clone(); - let command = options.command; - let cwd = options.cwd; - let env = options.env; - let user = options.user; - let timeout = options - .timeout - .map(ReadyTimeout::new) - .transpose() - .map_err(invalid_argument_status)?; - let background = options.background; - let stdin = options.stdin; - let pty = options.pty; - let clean_env = options.clean_env; - let assigned_process_id = options.process_id; - let idempotency_key = Some(match options.idempotency_key { - Some(key) => key, - None => IdempotencyKey::parse(uuid::Uuid::now_v7().to_string()) - .map_err(invalid_argument_status)?, - }); - pplx_space_spaced_common::retry_status( - "spaced host", - "process_start", - self.config.command_retry_policy, - || { - let mut client = self.process.clone(); - let command = Pii::new(&command).expose_pii().argv().to_vec(); - let cwd = cwd.as_ref().map(|cwd| cwd.expose().to_string()); - let env = proto_env(&env); - let user = user.as_ref().map(|user| user.as_str().to_string()); - let idempotency_key = idempotency_key.clone(); - let tag = tag.clone(); - let process_id = assigned_process_id - .as_ref() - .map(|process_id| process_id.to_string()); - async move { - client - .start(StartRequest { - command, - cwd, - env, - user, - timeout: timeout.map(proto_duration), - background, - idempotency_key: proto_idempotency_key(idempotency_key), - tag, - stdin, - pty, - clean_env, - process_id, - }) - .await - .map(Response::into_inner) - } - }, - ) - .await - } - - pub async fn connect_process_stream_with_cursor( - &self, - process_id: String, - timeout: Option, - output_cursor: Option, - ) -> Result, tonic::Status> { - let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; - let timeout = timeout - .map(ReadyTimeout::new) - .transpose() - .map_err(invalid_argument_status)?; - pplx_space_spaced_common::retry_status( - "spaced host", - "process_connect", - self.config.command_retry_policy, - || { - let mut client = self.process.clone(); - let process_id = process_id.clone(); - async move { - client - .connect_process(connect_process_request( - &process_id, - timeout, - output_cursor, - )) - .await - .map(Response::into_inner) - } - }, - ) - .await - } - - pub async fn list_processes(&self) -> Result, tonic::Status> { - tokio::time::timeout( - PROCESS_LIST_TIMEOUT, - pplx_space_spaced_common::retry_status( - "spaced host", - "process_list", - self.config.command_retry_policy, - || { - let mut client = self.process.clone(); - async move { - let mut stream = client - .list_processes(ListProcessesRequest {}) - .await? - .into_inner(); - let mut processes = Vec::new(); - let mut encoded_bytes = 0usize; - while let Some(process) = stream.message().await? { - if processes.len() == PROCESS_LIST_MAX_RECORDS { - return Err(tonic::Status::resource_exhausted( - "spaced host returned too many process records", - )); - } - encoded_bytes = encoded_bytes - .checked_add(process.encoded_len()) - .filter(|size| *size <= PROCESS_LIST_MAX_ENCODED_BYTES) - .ok_or_else(|| { - tonic::Status::resource_exhausted( - "spaced host process metadata is too large", - ) - })?; - processes.push(process); - } - Ok(processes) - } - }, - ), - ) - .await - .map_err(|_| tonic::Status::deadline_exceeded("spaced host process list timed out"))? - } - - /// Attempts a process signal without automatic replay. - /// - /// Use this for security-sensitive control paths where an ambiguous - /// transport failure must not replay a state-changing request. - pub async fn send_signal( - &self, - process_id: String, - signal: i32, - ) -> Result { - let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; - let signal = ProcessSignal::try_from_i32(signal).map_err(invalid_argument_status)?; - let mut client = self.process.clone(); - client - .send_signal(SendSignalRequest { - process_id: process_id.as_str().to_string(), - signal: signal.as_i32(), - }) - .await - .map(Response::into_inner) - } - - /// Attempts a process stdin write without replay so an ambiguous failure cannot - /// duplicate secret input. - pub async fn send_input( - &self, - process_id: String, - data: Vec, - ) -> Result { - let mut request = proto::SendInputRequest { process_id, data }; - let process_id = ProcessId::parse(std::mem::take(&mut request.process_id)) - .map_err(invalid_argument_status)?; - request.process_id = process_id.as_str().to_string(); - let mut client = self.process.clone(); - client.send_input(request).await.map(Response::into_inner) - } - - /// Attempts to close process stdin without replay so an ambiguous failure cannot - /// replay process state transitions. - pub async fn close_stdin( - &self, - process_id: String, - ) -> Result { - let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; - let mut client = self.process.clone(); - client - .close_stdin(proto::CloseStdinRequest { - process_id: process_id.as_str().to_string(), - }) - .await - .map(Response::into_inner) - } - - pub async fn resize_pty( - &self, - process_id: String, - size: proto::PtySize, - ) -> Result { - let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; - self.call_process("process_resize_pty", |mut client| { - let process_id = process_id.clone(); - async move { - client - .resize_pty(proto::ResizePtyRequest { - process_id: process_id.as_str().to_string(), - size: Some(size), - }) - .await - } - }) - .await - } - - /// Attempts to force-stop a process without replay. This is intended for bounded - /// cleanup after an earlier control request had an ambiguous outcome. - pub async fn kill_process( - &self, - process_id: String, - ) -> Result { - let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; - let mut client = self.process.clone(); - client - .kill(KillRequest { - process_id: process_id.as_str().to_string(), - }) - .await - .map(Response::into_inner) - } - - /// Open the in-VM loopback bridge for `port`, blocking server-side for - /// up to `wait_timeout` for the listener to appear. - /// - /// Bypasses the standard `command_retry_policy` because (a) the - /// service-side wait already absorbs transient "listener not yet up" - /// states by design, and (b) re-sending the same `wait_timeout` on - /// each retry would compound the budget (2s × N) without changing the - /// outcome — `WaitingForListener` is a "the port is armed; reconnect" - /// signal, not a transient transport error. - pub async fn open_loopback_bridge( - &self, - port: u32, - wait_timeout: Option, - ) -> Result { - let wait_timeout_proto = wait_timeout - .map(pplx_space_spaced_common::duration_to_proto) - .transpose()?; - let mut client = self.networking.clone(); - client - .open_loopback_bridge(proto::OpenLoopbackBridgeRequest { - port, - wait_timeout: wait_timeout_proto, - }) - .await - .map(Response::into_inner) - } - - /// Close authorization for `port` with a single dispatch. Client construction - /// retries connection establishment, but a lost close response is ambiguous: - /// replay could close a replacement bridge opened on the same port. - pub async fn close_loopback_bridge( - &self, - port: u32, - ) -> Result { - self.call_networking_once("close_loopback_bridge", move |mut client| async move { - client - .close_loopback_bridge(proto::CloseLoopbackBridgeRequest { port }) - .await - }) - .await - } } #[cfg(test)] diff --git a/crates/libs/spaced/host/client/src/meta.rs b/crates/libs/spaced/host/client/src/meta.rs new file mode 100644 index 000000000..39571c292 --- /dev/null +++ b/crates/libs/spaced/host/client/src/meta.rs @@ -0,0 +1,113 @@ +//! Status, readiness and lifecycle calls to the guest agent, including the process environment it +//! configures. + +use super::*; + +impl SpacedHostClient { + pub async fn ping(&self) -> Result { + self.call_meta("ping", |mut client| async move { + client.ping(PingRequest {}).await + }) + .await + } + + pub async fn is_connected(&self) -> Result { + self.call_meta("is_connected", |mut client| async move { + client.is_connected(IsConnectedRequest {}).await + }) + .await + } + + pub async fn ping_guest(&self) -> Result { + self.call_meta("ping_guest", |mut client| async move { + client.ping_guest(PingGuestRequest {}).await + }) + .await + } + + pub async fn get_status(&self) -> Result { + self.call_meta("get_status", |mut client| async move { + client.get_status(GetStatusRequest {}).await + }) + .await + } + + pub async fn wait_ready( + &self, + timeout: std::time::Duration, + ) -> Result { + let timeout = ReadyTimeout::new(timeout).map_err(invalid_argument_status)?; + self.call_meta("wait_ready", |mut client| async move { + client + .wait_ready(WaitReadyRequest { + timeout: Some(proto_duration(timeout)), + }) + .await + }) + .await + } + + pub async fn configure_process_env( + &self, + env: std::collections::HashMap, + ) -> Result<(), tonic::Status> { + let env = parse_env(env); + self.call_meta_once("configure_process_env", |mut client| { + let env = proto_env(&env); + async move { + client + .configure_process_env(ConfigureProcessEnvRequest { env }) + .await + } + }) + .await + .map(|_: proto::ConfigureProcessEnvResponse| ()) + } + + pub async fn notify_resume( + &self, + reason: ResumeReason, + ) -> Result { + let now = pplx_space_spaced_common::system_time_to_proto(std::time::SystemTime::now())?; + self.call_meta_once("notify_resume", |mut client| async move { + client + .notify_resume(NotifyResumeRequest { + reason: reason.into(), + host_time: Some(now), + }) + .await + }) + .await + } + + /// Prepare for a VM memory snapshot, including the final guest filesystem flush. + /// + /// Call this instead of `flush_guest_filesystems` for suspend or full + /// checkpoint paths; do not call both for one snapshot. + pub async fn prepare_suspend(&self, timeout: std::time::Duration) -> Result<(), tonic::Status> { + self.call_meta_once("prepare_suspend", |mut client| async move { + let mut request = tonic::Request::new(PrepareSuspendRequest {}); + request.set_timeout(timeout.min(std::time::Duration::from_secs(180))); + client.prepare_suspend(request).await + }) + .await + .map(|_: proto::PrepareSuspendResponse| ()) + } + + /// Flush guest filesystems without preparing bridge connections for a VM snapshot. + /// + /// Use this for disk-only snapshots of a still-running VM. Memory-bearing + /// snapshots must call `prepare_suspend` instead. + pub async fn flush_guest_filesystems( + &self, + timeout: std::time::Duration, + ) -> Result<(), tonic::Status> { + self.call_meta_once("flush_guest_filesystems", |mut client| async move { + let mut request = tonic::Request::new(FlushGuestFilesystemsRequest {}); + request.set_timeout(timeout.min(std::time::Duration::from_secs(180))); + client.flush_guest_filesystems(request).await + }) + .await + .map(|_: proto::FlushGuestFilesystemsResponse| ()) + } +} diff --git a/crates/libs/spaced/host/client/src/networking.rs b/crates/libs/spaced/host/client/src/networking.rs new file mode 100644 index 000000000..9e512dd2c --- /dev/null +++ b/crates/libs/spaced/host/client/src/networking.rs @@ -0,0 +1,47 @@ +//! Loopback bridges between the host and the guest. + +use super::*; + +impl SpacedHostClient { + /// Open the in-VM loopback bridge for `port`, blocking server-side for + /// up to `wait_timeout` for the listener to appear. + /// + /// Bypasses the standard `command_retry_policy` because (a) the + /// service-side wait already absorbs transient "listener not yet up" + /// states by design, and (b) re-sending the same `wait_timeout` on + /// each retry would compound the budget (2s × N) without changing the + /// outcome — `WaitingForListener` is a "the port is armed; reconnect" + /// signal, not a transient transport error. + pub async fn open_loopback_bridge( + &self, + port: u32, + wait_timeout: Option, + ) -> Result { + let wait_timeout_proto = wait_timeout + .map(pplx_space_spaced_common::duration_to_proto) + .transpose()?; + let mut client = self.networking.clone(); + client + .open_loopback_bridge(proto::OpenLoopbackBridgeRequest { + port, + wait_timeout: wait_timeout_proto, + }) + .await + .map(Response::into_inner) + } + + /// Close authorization for `port` with a single dispatch. Client construction + /// retries connection establishment, but a lost close response is ambiguous: + /// replay could close a replacement bridge opened on the same port. + pub async fn close_loopback_bridge( + &self, + port: u32, + ) -> Result { + self.call_networking_once("close_loopback_bridge", move |mut client| async move { + client + .close_loopback_bridge(proto::CloseLoopbackBridgeRequest { port }) + .await + }) + .await + } +} diff --git a/crates/libs/spaced/host/client/src/process.rs b/crates/libs/spaced/host/client/src/process.rs new file mode 100644 index 000000000..0061b3c8f --- /dev/null +++ b/crates/libs/spaced/host/client/src/process.rs @@ -0,0 +1,248 @@ +//! Process calls: start, reconnect, list, signals, input, PTY resize and kill. + +use super::*; + +pub(super) const PROCESS_LIST_MAX_RECORDS: usize = 128; + +pub(super) const PROCESS_LIST_MAX_ENCODED_BYTES: usize = 4 * 1024 * 1024; + +pub(super) const PROCESS_LIST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +pub(super) fn connect_process_request( + process_id: &ProcessId, + timeout: Option, + output_cursor: Option, +) -> ConnectProcessRequest { + ConnectProcessRequest { + process_id: process_id.as_str().to_string(), + timeout: timeout.map(proto_duration), + output_cursor, + } +} + +impl SpacedHostClient { + pub async fn start_process_stream_with_options( + &self, + options: StartProcessOptions, + ) -> Result, tonic::Status> { + let tag = options.tag.clone(); + let command = options.command; + let cwd = options.cwd; + let env = options.env; + let user = options.user; + let timeout = options + .timeout + .map(ReadyTimeout::new) + .transpose() + .map_err(invalid_argument_status)?; + let background = options.background; + let stdin = options.stdin; + let pty = options.pty; + let clean_env = options.clean_env; + let assigned_process_id = options.process_id; + let idempotency_key = Some(match options.idempotency_key { + Some(key) => key, + None => IdempotencyKey::parse(uuid::Uuid::now_v7().to_string()) + .map_err(invalid_argument_status)?, + }); + pplx_space_spaced_common::retry_status( + "spaced host", + "process_start", + self.config.command_retry_policy, + || { + let mut client = self.process.clone(); + let command = Pii::new(&command).expose_pii().argv().to_vec(); + let cwd = cwd.as_ref().map(|cwd| cwd.expose().to_string()); + let env = proto_env(&env); + let user = user.as_ref().map(|user| user.as_str().to_string()); + let idempotency_key = idempotency_key.clone(); + let tag = tag.clone(); + let process_id = assigned_process_id + .as_ref() + .map(|process_id| process_id.to_string()); + async move { + client + .start(StartRequest { + command, + cwd, + env, + user, + timeout: timeout.map(proto_duration), + background, + idempotency_key: proto_idempotency_key(idempotency_key), + tag, + stdin, + pty, + clean_env, + process_id, + }) + .await + .map(Response::into_inner) + } + }, + ) + .await + } + + pub async fn connect_process_stream_with_cursor( + &self, + process_id: String, + timeout: Option, + output_cursor: Option, + ) -> Result, tonic::Status> { + let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; + let timeout = timeout + .map(ReadyTimeout::new) + .transpose() + .map_err(invalid_argument_status)?; + pplx_space_spaced_common::retry_status( + "spaced host", + "process_connect", + self.config.command_retry_policy, + || { + let mut client = self.process.clone(); + let process_id = process_id.clone(); + async move { + client + .connect_process(connect_process_request( + &process_id, + timeout, + output_cursor, + )) + .await + .map(Response::into_inner) + } + }, + ) + .await + } + + pub async fn list_processes(&self) -> Result, tonic::Status> { + tokio::time::timeout( + PROCESS_LIST_TIMEOUT, + pplx_space_spaced_common::retry_status( + "spaced host", + "process_list", + self.config.command_retry_policy, + || { + let mut client = self.process.clone(); + async move { + let mut stream = client + .list_processes(ListProcessesRequest {}) + .await? + .into_inner(); + let mut processes = Vec::new(); + let mut encoded_bytes = 0usize; + while let Some(process) = stream.message().await? { + if processes.len() == PROCESS_LIST_MAX_RECORDS { + return Err(tonic::Status::resource_exhausted( + "spaced host returned too many process records", + )); + } + encoded_bytes = encoded_bytes + .checked_add(process.encoded_len()) + .filter(|size| *size <= PROCESS_LIST_MAX_ENCODED_BYTES) + .ok_or_else(|| { + tonic::Status::resource_exhausted( + "spaced host process metadata is too large", + ) + })?; + processes.push(process); + } + Ok(processes) + } + }, + ), + ) + .await + .map_err(|_| tonic::Status::deadline_exceeded("spaced host process list timed out"))? + } + + /// Attempts a process signal without automatic replay. + /// + /// Use this for security-sensitive control paths where an ambiguous + /// transport failure must not replay a state-changing request. + pub async fn send_signal( + &self, + process_id: String, + signal: i32, + ) -> Result { + let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; + let signal = ProcessSignal::try_from_i32(signal).map_err(invalid_argument_status)?; + let mut client = self.process.clone(); + client + .send_signal(SendSignalRequest { + process_id: process_id.as_str().to_string(), + signal: signal.as_i32(), + }) + .await + .map(Response::into_inner) + } + + /// Attempts a process stdin write without replay so an ambiguous failure cannot + /// duplicate secret input. + pub async fn send_input( + &self, + process_id: String, + data: Vec, + ) -> Result { + let mut request = proto::SendInputRequest { process_id, data }; + let process_id = ProcessId::parse(std::mem::take(&mut request.process_id)) + .map_err(invalid_argument_status)?; + request.process_id = process_id.as_str().to_string(); + let mut client = self.process.clone(); + client.send_input(request).await.map(Response::into_inner) + } + + /// Attempts to close process stdin without replay so an ambiguous failure cannot + /// replay process state transitions. + pub async fn close_stdin( + &self, + process_id: String, + ) -> Result { + let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; + let mut client = self.process.clone(); + client + .close_stdin(proto::CloseStdinRequest { + process_id: process_id.as_str().to_string(), + }) + .await + .map(Response::into_inner) + } + + pub async fn resize_pty( + &self, + process_id: String, + size: proto::PtySize, + ) -> Result { + let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; + self.call_process("process_resize_pty", |mut client| { + let process_id = process_id.clone(); + async move { + client + .resize_pty(proto::ResizePtyRequest { + process_id: process_id.as_str().to_string(), + size: Some(size), + }) + .await + } + }) + .await + } + + /// Attempts to force-stop a process without replay. This is intended for bounded + /// cleanup after an earlier control request had an ambiguous outcome. + pub async fn kill_process( + &self, + process_id: String, + ) -> Result { + let process_id = ProcessId::parse(process_id).map_err(invalid_argument_status)?; + let mut client = self.process.clone(); + client + .kill(KillRequest { + process_id: process_id.as_str().to_string(), + }) + .await + .map(Response::into_inner) + } +}