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) + } +}