1use std::sync::Arc;
30use std::time::Duration;
31use time::OffsetDateTime;
32
33use crate::app::{self, App};
34use crate::config::Config;
35use crate::credential::api::CredentialApi;
36use crate::credential::types::{
37 CredentialInjectionPolicyInfo, CredentialInjectionPolicyPage, InjectionRule,
38 ListCredentialInjectionPoliciesQuery, SecretInfo,
39};
40use crate::error::{RpcStatus, SailError};
41use crate::exec::{ExecOptions, ExecParams, ExecProcess, ExecResult, OutputStream};
42use crate::http::HttpCore;
43use crate::imagebuild::BuildMode;
44use crate::imagebuilder::ImageBuilder;
45use crate::sailbox::api::{SailboxApi, UpgradeResult};
46use crate::sailbox::fs::{DirEntry, EntryType};
47use crate::sailbox::object::Sailbox;
48use crate::sailbox::types::{
49 CreateSailboxRequest, ListSailboxesQuery, SailboxCheckpoint, SailboxHandle, SailboxInfo,
50 SailboxMetricsQuery, SailboxMetricsResponse, SailboxPage, SailboxSpendQuery,
51 SailboxSpendResponse, VolumeInfo, WhoAmI,
52};
53use crate::worker::{
54 is_transient_transport_message, FileReader, FileWriter, Listener, WorkerProxy, WriteOptions,
55};
56
57#[derive(Clone)]
59pub struct Client {
60 inner: Arc<Inner>,
61}
62
63impl std::fmt::Debug for Client {
64 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65 f.debug_struct("Client")
66 .field("config", &self.inner.config)
67 .finish_non_exhaustive()
68 }
69}
70
71struct Inner {
72 config: Config,
73 sailbox_http: HttpCore,
75 api_http: HttpCore,
77 worker: Arc<WorkerProxy>,
80 imagebuilder: ImageBuilder,
81 image_ready: crate::imagecache::ImageReadyCache,
84}
85
86const HINTED_EXEC_ENDPOINT_PROBE_TIMEOUT: Duration = Duration::from_secs(1);
89
90#[derive(Default, Clone)]
96pub struct ClientBuilder {
97 mode: Option<String>,
98 api_key: Option<String>,
99 api_url: Option<String>,
100 sailbox_api_url: Option<String>,
101 imagebuilder_url: Option<String>,
102 ingress_url: Option<String>,
103 client_label: Option<String>,
104}
105
106impl std::fmt::Debug for ClientBuilder {
107 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
108 f.debug_struct("ClientBuilder")
109 .field(
110 "api_key",
111 &crate::config::redact_key(self.api_key.as_deref().unwrap_or("")),
112 )
113 .field("mode", &self.mode)
114 .field("api_url", &self.api_url)
115 .field("sailbox_api_url", &self.sailbox_api_url)
116 .field("imagebuilder_url", &self.imagebuilder_url)
117 .field("ingress_url", &self.ingress_url)
118 .field("client_label", &self.client_label)
119 .finish()
120 }
121}
122
123impl ClientBuilder {
124 pub fn new(api_key: impl Into<String>) -> ClientBuilder {
127 ClientBuilder {
128 api_key: Some(api_key.into()),
129 ..ClientBuilder::default()
130 }
131 }
132
133 #[doc(hidden)]
136 pub fn mode(mut self, mode: impl Into<String>) -> ClientBuilder {
137 self.mode = Some(mode.into());
138 self
139 }
140
141 pub fn api_url(mut self, api_url: impl Into<String>) -> ClientBuilder {
143 self.api_url = Some(api_url.into());
144 self
145 }
146
147 pub fn sailbox_api_url(mut self, url: impl Into<String>) -> ClientBuilder {
149 self.sailbox_api_url = Some(url.into());
150 self
151 }
152
153 pub fn imagebuilder_url(mut self, url: impl Into<String>) -> ClientBuilder {
155 self.imagebuilder_url = Some(url.into());
156 self
157 }
158
159 pub fn ingress_url(mut self, url: impl Into<String>) -> ClientBuilder {
162 self.ingress_url = Some(url.into());
163 self
164 }
165
166 #[doc(hidden)]
168 pub fn client_label(mut self, label: impl Into<String>) -> ClientBuilder {
169 self.client_label = Some(label.into());
170 self
171 }
172
173 pub fn build(self) -> Result<Client, SailError> {
175 let api_key = self.api_key.unwrap_or_default();
176 let config = Config::resolve(
177 self.mode.as_deref(),
178 api_key,
179 self.api_url,
180 self.sailbox_api_url,
181 self.imagebuilder_url,
182 self.ingress_url,
183 )?;
184 Client::from_config_with_label(
185 config,
186 self.client_label
187 .as_deref()
188 .unwrap_or(crate::http::DEFAULT_CLIENT_LABEL),
189 )
190 }
191}
192
193const STALE_IMAGE_REBUILD_TIMEOUT: Duration = Duration::from_mins(30);
197
198fn image_not_ready_conflict(result: &Result<SailboxHandle, SailError>) -> bool {
202 matches!(
203 result,
204 Err(SailError::Creation {
205 status: 409,
206 message,
207 ..
208 }) if message.starts_with("resolve image:")
209 )
210}
211
212impl Client {
213 pub fn builder(api_key: impl Into<String>) -> ClientBuilder {
215 ClientBuilder::new(api_key)
216 }
217
218 pub fn from_env() -> Result<Client, SailError> {
220 Client::from_config(Config::from_env()?)
221 }
222
223 #[doc(hidden)]
225 pub fn from_env_with_label(label: &str) -> Result<Client, SailError> {
226 Client::from_config_with_label(Config::from_env()?, label)
227 }
228
229 pub fn from_config(config: Config) -> Result<Client, SailError> {
231 Client::from_config_with_label(config, crate::http::DEFAULT_CLIENT_LABEL)
232 }
233
234 fn from_config_with_label(config: Config, client_label: &str) -> Result<Client, SailError> {
235 let sailbox_http = HttpCore::new(&config.sailbox_api_url, &config.api_key)?
236 .with_client_label(client_label);
237 let api_http =
238 HttpCore::new(&config.api_url, &config.api_key)?.with_client_label(client_label);
239 let worker = Arc::new(WorkerProxy::new(&config.api_key)?);
240 let imagebuilder = ImageBuilder::new(&config.imagebuilder_url, &config.api_key)?;
241 Ok(Client {
242 inner: Arc::new(Inner {
243 config,
244 sailbox_http,
245 api_http,
246 worker,
247 imagebuilder,
248 image_ready: crate::imagecache::ImageReadyCache::new(),
249 }),
250 })
251 }
252
253 pub(crate) fn image_ready_cache(&self) -> &crate::imagecache::ImageReadyCache {
254 &self.inner.image_ready
255 }
256
257 #[cfg(any(test, feature = "test-fakes"))]
262 pub fn set_image_ready_refresh_window(&self, window: std::time::Duration) {
263 self.inner.image_ready.set_refresh_window(window);
264 }
265
266 pub fn config(&self) -> &Config {
268 &self.inner.config
269 }
270
271 #[doc(hidden)]
273 pub fn worker(&self) -> Arc<WorkerProxy> {
274 Arc::clone(&self.inner.worker)
275 }
276
277 #[doc(hidden)]
279 pub fn imagebuilder(&self) -> &ImageBuilder {
280 &self.inner.imagebuilder
281 }
282
283 #[doc(hidden)]
285 pub fn sailbox_http(&self) -> &HttpCore {
286 &self.inner.sailbox_http
287 }
288
289 #[doc(hidden)]
291 pub fn api_http(&self) -> &HttpCore {
292 &self.inner.api_http
293 }
294
295 fn sailbox_api(&self) -> SailboxApi<'_> {
296 SailboxApi::new(&self.inner.sailbox_http)
297 }
298
299 async fn create_with_image_revalidation(
307 &self,
308 req: &CreateSailboxRequest,
309 timeout: Option<Duration>,
310 ) -> Result<SailboxHandle, SailError> {
311 let create_started = std::time::Instant::now();
312 let result = self.sailbox_api().create(req, timeout).await;
313 let custom_image = req.image != crate::image::ImageSpec::default()
314 && !crate::imagebuild::is_builtin_base_spec(&req.image);
315 if !custom_image || !image_not_ready_conflict(&result) {
316 return result;
317 }
318 if let Ok(spec_hash) = crate::imagebuild::canonical_spec_key(&req.image) {
319 self.image_ready_cache()
324 .invalidate_spec_started_before(&spec_hash, create_started);
325 }
326 let rebuild_timeout = req
329 .image_build_timeout
330 .unwrap_or(STALE_IMAGE_REBUILD_TIMEOUT);
331 let rebuild = self.build_spec_ready_cached(
332 &req.image,
333 rebuild_timeout,
334 crate::imagecache::BuildOrigin::StaleCreateRecovery,
335 BuildMode::ReuseExisting,
336 );
337 let build = tokio::time::timeout(rebuild_timeout, rebuild)
338 .await
339 .unwrap_or_else(|_| {
340 Err(SailError::Transport {
341 kind: crate::error::TransportKind::Timeout,
342 message: "timed out building the image".to_string(),
343 source: None,
344 })
345 })?;
346 let mut retry = req.clone();
352 crate::imagebuild::pin_resolved_oci_ref(&mut retry.image, &build.resolved_oci_ref);
353 crate::imagebuild::pin_dockerfile_from(&mut retry.image, build.dockerfile_pins.as_deref());
354 self.sailbox_api().create(&retry, timeout).await
355 }
356
357 pub async fn create_sailbox(
369 &self,
370 req: &CreateSailboxRequest,
371 timeout: Option<Duration>,
372 ) -> Result<Sailbox, SailError> {
373 crate::imagebuild::validate_image_spec_source(&req.image)?;
378 let bind = |handle: SailboxHandle| Sailbox::bind(self.clone(), handle);
379 if !req.ssh {
380 return self
381 .create_with_image_revalidation(req, timeout)
382 .await
383 .map(bind);
384 }
385 crate::sailbox::api::validate_ingress_ports(&req.ingress_ports)?;
390 self.org_ssh_ca_public_key().await?;
393 let mut req = req.clone();
398 let ssh_allowlist = req
399 .ingress_ports
400 .iter()
401 .find(|port| port.guest_port == 22)
402 .map(|port| port.allowlist.clone())
403 .unwrap_or_default();
404 req.ingress_ports.retain(|port| port.guest_port != 22);
405 let handle = self.create_with_image_revalidation(&req, timeout).await?;
406 let handle_id = handle.sailbox_id.clone();
407 if let Err(err) = self
409 .enable_ssh(
410 &handle_id,
411 &ssh_allowlist,
412 false,
413 Duration::ZERO,
414 )
415 .await
416 {
417 return Err(SailError::Creation {
420 message: format!(
421 "sailbox {handle_id} was created, but SSH setup failed: {err}. Fetch it by \
422 id to retry enable_ssh or terminate it."
423 ),
424 status: 0,
425 body: serde_json::Value::Null,
426 });
427 }
428 Ok(bind(handle))
429 }
430
431 #[doc(hidden)]
433 pub async fn get_sailbox(&self, sailbox_id: &str) -> Result<SailboxInfo, SailError> {
434 self.sailbox_api().get(sailbox_id).await
435 }
436
437 #[doc(hidden)]
439 pub async fn whoami(&self) -> Result<WhoAmI, SailError> {
440 self.sailbox_api().whoami().await
441 }
442
443 pub async fn list_sailboxes(
445 &self,
446 query: &ListSailboxesQuery,
447 ) -> Result<SailboxPage, SailError> {
448 self.sailbox_api().list(query).await
449 }
450
451 pub async fn sailbox_spend(
453 &self,
454 query: &SailboxSpendQuery,
455 ) -> Result<SailboxSpendResponse, SailError> {
456 self.sailbox_api().spend(query).await
457 }
458
459 pub async fn sailbox_metrics(
461 &self,
462 sailbox_id: &str,
463 query: &SailboxMetricsQuery,
464 ) -> Result<SailboxMetricsResponse, SailError> {
465 self.sailbox_api().metrics(sailbox_id, query).await
466 }
467
468 #[doc(hidden)]
470 pub async fn terminate_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
471 self.sailbox_api().terminate(sailbox_id).await
472 }
473
474 #[doc(hidden)]
476 pub async fn pause_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
477 self.sailbox_api().pause(sailbox_id).await
478 }
479
480 #[doc(hidden)]
482 pub async fn sleep_sailbox(
483 &self,
484 sailbox_id: &str,
485 wake_at: Option<OffsetDateTime>,
486 ) -> Result<Option<OffsetDateTime>, SailError> {
487 self.sailbox_api().sleep(sailbox_id, wake_at).await
488 }
489
490 #[doc(hidden)]
492 pub async fn set_sailbox_auto_sleep(
493 &self,
494 sailbox_id: &str,
495 auto_sleep: crate::AutoSleep,
496 ) -> Result<(), SailError> {
497 self.sailbox_api()
498 .set_auto_sleep(sailbox_id, auto_sleep)
499 .await
500 }
501
502 #[doc(hidden)]
504 pub async fn resume_sailbox(&self, sailbox_id: &str) -> Result<SailboxHandle, SailError> {
505 self.sailbox_api().resume(sailbox_id).await
506 }
507
508 #[doc(hidden)]
510 pub async fn checkpoint_sailbox(
511 &self,
512 sailbox_id: &str,
513 name: Option<&str>,
514 ttl_seconds: Option<i64>,
515 ) -> Result<SailboxCheckpoint, SailError> {
516 self.sailbox_api()
517 .checkpoint(sailbox_id, name, ttl_seconds)
518 .await
519 }
520
521 pub async fn create_from_checkpoint(
541 &self,
542 checkpoint_id: &str,
543 name: Option<&str>,
544 timeout: Option<Duration>,
545 ) -> Result<Sailbox, SailError> {
546 self.sailbox_api()
547 .from_checkpoint(checkpoint_id, name, timeout.map(duration_to_whole_seconds))
548 .await
549 .map(|handle| Sailbox::bind(self.clone(), handle))
550 }
551
552 #[doc(hidden)]
554 pub async fn upgrade_sailbox(&self, sailbox_id: &str) -> Result<UpgradeResult, SailError> {
555 self.sailbox_api().upgrade(sailbox_id).await
556 }
557
558 #[doc(hidden)]
563 pub async fn expose_listener(
564 &self,
565 sailbox_id: &str,
566 guest_port: u32,
567 protocol: crate::sailbox::types::IngressProtocol,
568 allowlist: &[String],
569 ) -> Result<Listener, SailError> {
570 let mut response = self
571 .sailbox_api()
572 .expose(sailbox_id, guest_port, protocol, allowlist)
573 .await?;
574 self.fill_listener_url(sailbox_id, &mut response);
575 Ok(response)
576 }
577
578 #[doc(hidden)]
580 pub async fn unexpose_listener(
581 &self,
582 sailbox_id: &str,
583 guest_port: u32,
584 ) -> Result<(), SailError> {
585 self.sailbox_api().unexpose(sailbox_id, guest_port).await
586 }
587
588 #[doc(hidden)]
590 pub async fn list_listeners(
591 &self,
592 sailbox_id: &str,
593 ) -> Result<Vec<crate::worker::Listener>, SailError> {
594 let mut listeners = self.sailbox_api().list_listeners(sailbox_id).await?;
595 for listener in &mut listeners {
596 self.fill_listener_url(sailbox_id, listener);
597 }
598 Ok(listeners)
599 }
600
601 #[doc(hidden)]
604 pub async fn get_listener(
605 &self,
606 sailbox_id: &str,
607 guest_port: u32,
608 ) -> Result<crate::worker::Listener, SailError> {
609 let mut listener = self
610 .sailbox_api()
611 .get_listener(sailbox_id, guest_port)
612 .await?;
613 self.fill_listener_url(sailbox_id, &mut listener);
614 Ok(listener)
615 }
616
617 #[doc(hidden)]
619 pub async fn custom_domain_dns_targets(&self) -> Result<(String, Option<String>), SailError> {
620 self.sailbox_api().custom_domain_dns_targets().await
621 }
622
623 #[doc(hidden)]
625 pub async fn attach_custom_domain(
626 &self,
627 sailbox_id: &str,
628 domain: &str,
629 guest_port: u32,
630 ) -> Result<crate::sailbox::types::CustomDomainInfo, SailError> {
631 self.sailbox_api()
632 .attach_custom_domain(sailbox_id, domain, guest_port)
633 .await
634 }
635
636 #[doc(hidden)]
638 pub async fn list_custom_domains(
639 &self,
640 sailbox_id: &str,
641 ) -> Result<Vec<crate::sailbox::types::CustomDomainInfo>, SailError> {
642 self.sailbox_api().list_custom_domains(sailbox_id).await
643 }
644
645 #[doc(hidden)]
647 pub async fn detach_custom_domain(
648 &self,
649 sailbox_id: &str,
650 domain: &str,
651 ) -> Result<(), SailError> {
652 self.sailbox_api()
653 .detach_custom_domain(sailbox_id, domain)
654 .await
655 }
656
657 fn fill_listener_url(&self, sailbox_id: &str, listener: &mut crate::worker::Listener) {
661 if listener.public_url.is_empty()
662 && listener.protocol != crate::sailbox::types::ListenerProtocol::Tcp
663 {
664 listener.public_url = crate::sailbox::listeners::synthesized_public_url(
665 self.config(),
666 sailbox_id,
667 listener.guest_port,
668 );
669 }
670 }
671
672 #[doc(hidden)]
674 pub async fn ingress_auth_headers(
675 &self,
676 sailbox_id: &str,
677 ) -> Result<Vec<(String, String)>, SailError> {
678 self.sailbox_api().ingress_auth_headers(sailbox_id).await
679 }
680
681 pub async fn org_ssh_ca_public_key(&self) -> Result<String, SailError> {
683 self.sailbox_api().org_ssh_ca_public_key().await
684 }
685
686 pub async fn issue_user_cert(
690 &self,
691 public_key: &str,
692 timeout: Option<Duration>,
693 ) -> Result<crate::sailbox::types::IssuedUserCert, SailError> {
694 self.sailbox_api()
695 .issue_user_cert(public_key, timeout.map(|t| t.as_secs_f64()))
696 .await
697 }
698
699 pub async fn get_volume(
703 &self,
704 name: &str,
705 mint_if_missing: bool,
706 ) -> Result<VolumeInfo, SailError> {
707 self.sailbox_api().get_volume(name, mint_if_missing).await
708 }
709
710 pub async fn list_volumes(
712 &self,
713 max_objects: Option<i64>,
714 ) -> Result<Vec<VolumeInfo>, SailError> {
715 self.sailbox_api().list_volumes(max_objects).await
716 }
717
718 pub async fn delete_volume(
720 &self,
721 volume_id: &str,
722 allow_missing: bool,
723 ) -> Result<Option<VolumeInfo>, SailError> {
724 self.sailbox_api()
725 .delete_volume(volume_id, allow_missing)
726 .await
727 }
728
729 fn credential_api(&self) -> CredentialApi<'_> {
736 CredentialApi::new(&self.inner.sailbox_http)
737 }
738
739 #[doc(hidden)]
741 pub async fn set_secret(&self, name: &str, value: &str) -> Result<SecretInfo, SailError> {
742 self.credential_api().set_secret(name, value).await
743 }
744
745 #[doc(hidden)]
747 pub async fn get_secret(&self, name: &str) -> Result<SecretInfo, SailError> {
748 self.credential_api().get_secret(name).await
749 }
750
751 #[doc(hidden)]
753 pub async fn list_secrets(&self) -> Result<Vec<SecretInfo>, SailError> {
754 self.credential_api().list_secrets().await
755 }
756
757 #[doc(hidden)]
759 pub async fn delete_secret(&self, name: &str) -> Result<(), SailError> {
760 self.credential_api().delete_secret(name).await
761 }
762
763 #[doc(hidden)]
765 pub async fn create_credential_policy(
766 &self,
767 name: &str,
768 rules: &[InjectionRule],
769 ) -> Result<CredentialInjectionPolicyInfo, SailError> {
770 self.credential_api().create_policy(name, rules).await
771 }
772
773 #[doc(hidden)]
775 pub async fn get_credential_policy(
776 &self,
777 policy_id: &str,
778 ) -> Result<CredentialInjectionPolicyInfo, SailError> {
779 self.credential_api().get_policy(policy_id).await
780 }
781
782 #[doc(hidden)]
784 pub async fn list_credential_policies(
785 &self,
786 query: &ListCredentialInjectionPoliciesQuery,
787 ) -> Result<CredentialInjectionPolicyPage, SailError> {
788 self.credential_api().list_policies(query).await
789 }
790
791 #[doc(hidden)]
793 pub async fn rename_credential_policy(
794 &self,
795 policy_id: &str,
796 name: &str,
797 ) -> Result<CredentialInjectionPolicyInfo, SailError> {
798 self.credential_api().rename_policy(policy_id, name).await
799 }
800
801 #[doc(hidden)]
803 pub async fn delete_credential_policy(&self, policy_id: &str) -> Result<(), SailError> {
804 self.credential_api().delete_policy(policy_id).await
805 }
806
807 #[doc(hidden)]
809 pub async fn sailbox_credential_policy(
810 &self,
811 sailbox_id: &str,
812 ) -> Result<Option<CredentialInjectionPolicyInfo>, SailError> {
813 self.credential_api().sailbox_policy(sailbox_id).await
814 }
815
816 #[doc(hidden)]
818 pub async fn set_sailbox_credential_policy(
819 &self,
820 sailbox_id: &str,
821 policy_id: &str,
822 ) -> Result<(), SailError> {
823 self.credential_api()
824 .attach_sailbox_policy(sailbox_id, policy_id)
825 .await
826 }
827
828 #[doc(hidden)]
830 pub async fn clear_sailbox_credential_policy(&self, sailbox_id: &str) -> Result<(), SailError> {
831 self.credential_api()
832 .detach_sailbox_policy(sailbox_id)
833 .await
834 }
835
836 pub async fn find_app(&self, name: &str, mint_if_missing: bool) -> Result<App, SailError> {
840 app::find_app(&self.inner.api_http, name, mint_if_missing).await
841 }
842
843 pub async fn list_apps(&self) -> Result<Vec<App>, SailError> {
845 app::list_apps(&self.inner.api_http).await
846 }
847
848 #[doc(hidden)]
858 pub async fn exec_endpoint(&self, sailbox_id: &str) -> Result<String, SailError> {
859 let handle = self.resume_sailbox(sailbox_id).await?;
860 if handle.exec_endpoint.is_empty() {
861 return Err(SailError::Internal {
862 message: format!("sailbox {sailbox_id} resumed without an exec endpoint"),
863 });
864 }
865 Ok(handle.exec_endpoint)
866 }
867
868 #[doc(hidden)]
871 pub async fn exec(
872 &self,
873 sailbox_id: &str,
874 argv: Vec<String>,
875 options: ExecOptions,
876 ) -> Result<ExecProcess, SailError> {
877 self.exec_at_endpoint(sailbox_id, None, argv, options).await
878 }
879
880 #[doc(hidden)]
884 pub async fn exec_at_endpoint(
885 &self,
886 sailbox_id: &str,
887 exec_endpoint: Option<&str>,
888 argv: Vec<String>,
889 options: ExecOptions,
890 ) -> Result<ExecProcess, SailError> {
891 if argv.is_empty() {
892 return Err(SailError::InvalidArgument {
893 message: "command must be non-empty".to_string(),
894 });
895 }
896 if options.cwd.is_some() || options.background {
897 return Err(SailError::InvalidArgument {
898 message: "cwd and background require a shell command; use exec_shell or run_shell"
899 .to_string(),
900 });
901 }
902 let env = crate::exec::encode_env(&options.env)?;
905 let hinted_endpoint = exec_endpoint.filter(|endpoint| !endpoint.is_empty());
906 let exec_endpoint = match hinted_endpoint {
907 Some(endpoint) => endpoint.to_string(),
908 None => self.exec_endpoint(sailbox_id).await?,
909 };
910 let params = ExecParams {
911 sailbox_id: sailbox_id.to_string(),
912 exec_endpoint,
913 argv,
914 timeout_seconds: options
917 .timeout
918 .map_or(0, |d| d.as_secs_f64().ceil().max(1.0) as u32),
919 idempotency_key: options.idempotency_key,
920 open_stdin: options.open_stdin || options.pty,
923 pty: options.pty,
924 term: options.term,
925 cols: options.cols,
926 rows: options.rows,
927 env,
928 retry_timeout: options.retry_timeout.as_secs_f64(),
929 forward_ports: options.forward_ports,
930 forward_browser: options.forward_browser,
931 extra_metadata: Vec::new(),
932 forward_clipboard: options.forward_clipboard && options.pty,
935 user: options.user.unwrap_or_default(),
936 };
937 self.start_exec_params_at_endpoint(params, hinted_endpoint.is_some())
938 .await
939 }
940
941 #[doc(hidden)]
945 pub async fn start_exec_params_at_endpoint(
946 &self,
947 mut params: ExecParams,
948 endpoint_was_hint: bool,
949 ) -> Result<ExecProcess, SailError> {
950 if !endpoint_was_hint {
951 return ExecProcess::start(self.worker(), params).await;
952 }
953
954 params.ensure_idempotency_key();
961 let hinted_start =
962 ExecProcess::start_with_initial_retry_timeout(self.worker(), params.clone(), Some(0.0));
963 match tokio::time::timeout(HINTED_EXEC_ENDPOINT_PROBE_TIMEOUT, hinted_start).await {
964 Ok(Ok(process)) => return Ok(process),
965 Ok(Err(err)) if !should_reresolve_hinted_exec_endpoint(&err) => return Err(err),
966 Ok(Err(_)) | Err(_) => {}
967 }
968
969 self.worker().channels().invalidate(¶ms.exec_endpoint);
975 let endpoint = self.exec_endpoint(¶ms.sailbox_id).await?;
976 params.exec_endpoint = endpoint;
977 ExecProcess::start(self.worker(), params).await
978 }
979
980 #[doc(hidden)]
984 pub async fn exec_shell(
985 &self,
986 sailbox_id: &str,
987 command: &str,
988 options: ExecOptions,
989 ) -> Result<ExecProcess, SailError> {
990 self.exec_shell_at_endpoint(sailbox_id, None, command, options)
991 .await
992 }
993
994 #[doc(hidden)]
996 pub async fn exec_shell_at_endpoint(
997 &self,
998 sailbox_id: &str,
999 exec_endpoint: Option<&str>,
1000 command: &str,
1001 mut options: ExecOptions,
1002 ) -> Result<ExecProcess, SailError> {
1003 let argv = crate::exec::shell_argv(command, &options)?;
1004 options.cwd = None;
1007 options.background = false;
1008 self.exec_at_endpoint(sailbox_id, exec_endpoint, argv, options)
1009 .await
1010 }
1011
1012 #[doc(hidden)]
1020 pub async fn read_stream(
1021 &self,
1022 sailbox_id: &str,
1023 remote_path: &str,
1024 ) -> Result<FileReader, SailError> {
1025 let endpoint = self.exec_endpoint(sailbox_id).await?;
1026 Ok(self
1027 .inner
1028 .worker
1029 .read_file(&endpoint, sailbox_id, remote_path))
1030 }
1031
1032 #[doc(hidden)]
1036 pub async fn read_file(
1037 &self,
1038 sailbox_id: &str,
1039 remote_path: &str,
1040 ) -> Result<Vec<u8>, SailError> {
1041 let reader = self.read_stream(sailbox_id, remote_path).await?;
1042 let mut contents = Vec::new();
1043 while let Some(chunk) = reader.next().await {
1044 contents.extend_from_slice(&chunk?);
1045 }
1046 Ok(contents)
1047 }
1048
1049 #[doc(hidden)]
1058 pub async fn write_stream(
1059 &self,
1060 sailbox_id: &str,
1061 remote_path: &str,
1062 options: WriteOptions,
1063 ) -> Result<FileWriter, SailError> {
1064 let endpoint = self.exec_endpoint(sailbox_id).await?;
1065 Ok(self.inner.worker.write_file(
1066 &endpoint,
1067 sailbox_id,
1068 remote_path,
1069 options.create_parents,
1070 options.mode,
1071 options.user,
1072 ))
1073 }
1074
1075 #[doc(hidden)]
1079 pub async fn write_file(
1080 &self,
1081 sailbox_id: &str,
1082 remote_path: &str,
1083 data: &[u8],
1084 options: WriteOptions,
1085 ) -> Result<(), SailError> {
1086 let mut writer = self.write_stream(sailbox_id, remote_path, options).await?;
1087 writer.write(data).await?;
1088 writer.finish().await
1089 }
1090
1091 fn fs_exec_options(user: Option<String>) -> ExecOptions {
1114 ExecOptions {
1115 user: Some(
1116 user.filter(|u| !u.is_empty())
1117 .unwrap_or_else(|| "0:0".to_string()),
1118 ),
1119 ..ExecOptions::default()
1120 }
1121 }
1122
1123 async fn run_argv(
1125 &self,
1126 sailbox_id: &str,
1127 argv: Vec<String>,
1128 user: Option<String>,
1129 ) -> Result<ExecResult, SailError> {
1130 self.exec(sailbox_id, argv, Self::fs_exec_options(user))
1131 .await?
1132 .wait()
1133 .await
1134 }
1135
1136 #[doc(hidden)]
1141 pub async fn make_dir(
1142 &self,
1143 sailbox_id: &str,
1144 path: &str,
1145 user: Option<String>,
1146 ) -> Result<(), SailError> {
1147 crate::sailbox::fs::require_path(path)?;
1148 let result = self
1149 .run_argv(
1150 sailbox_id,
1151 vec![
1152 "mkdir".to_string(),
1153 "-p".to_string(),
1154 "--".to_string(),
1155 path.to_string(),
1156 ],
1157 user,
1158 )
1159 .await?;
1160 fs_command_ok(&result, &format!("create directory {path}"))
1161 }
1162
1163 #[doc(hidden)]
1167 pub async fn remove_path(
1168 &self,
1169 sailbox_id: &str,
1170 path: &str,
1171 user: Option<String>,
1172 ) -> Result<(), SailError> {
1173 crate::sailbox::fs::require_path(path)?;
1174 let result = self
1175 .run_argv(
1176 sailbox_id,
1177 vec![
1178 "rm".to_string(),
1179 "-rf".to_string(),
1180 "--".to_string(),
1181 path.to_string(),
1182 ],
1183 user,
1184 )
1185 .await?;
1186 fs_command_ok(&result, &format!("remove {path}"))
1187 }
1188
1189 #[doc(hidden)]
1194 pub async fn path_exists(
1195 &self,
1196 sailbox_id: &str,
1197 path: &str,
1198 user: Option<String>,
1199 ) -> Result<bool, SailError> {
1200 crate::sailbox::fs::require_path(path)?;
1201 let result = self
1202 .run_argv(
1203 sailbox_id,
1204 vec!["test".to_string(), "-e".to_string(), path.to_string()],
1205 user,
1206 )
1207 .await?;
1208 match result.exit_code {
1212 0 => Ok(true),
1213 1 => Ok(false),
1214 _ => Err(fs_command_error(
1215 &result,
1216 &format!("check whether {path} exists"),
1217 )),
1218 }
1219 }
1220
1221 #[doc(hidden)]
1227 pub async fn list_dir(
1228 &self,
1229 sailbox_id: &str,
1230 path: &str,
1231 user: Option<String>,
1232 ) -> Result<Vec<DirEntry>, SailError> {
1233 crate::sailbox::fs::require_path(path)?;
1234 let process = self
1235 .exec(
1236 sailbox_id,
1237 crate::sailbox::fs::list_dir_argv(path),
1238 Self::fs_exec_options(user),
1239 )
1240 .await?;
1241 let result = process.wait().await?;
1242 fs_command_ok(&result, &format!("list directory {path}"))?;
1243 if result.stdout_truncated {
1246 return Err(SailError::Execution {
1247 code: RpcStatus::FailedPrecondition,
1248 detail: format!(
1249 "directory listing for {path} was truncated because it has \
1250 too many entries; list a smaller subtree"
1251 ),
1252 });
1253 }
1254 if !result.stdout_complete {
1260 return Err(SailError::Execution {
1261 code: RpcStatus::FailedPrecondition,
1262 detail: format!(
1263 "directory listing for {path} was interrupted before it \
1264 finished streaming; retry the call"
1265 ),
1266 });
1267 }
1268 let mut entries =
1269 crate::sailbox::fs::parse_dir_entries(&process.buffered_output(OutputStream::Stdout))
1270 .map_err(|detail| SailError::Execution {
1271 code: RpcStatus::FailedPrecondition,
1272 detail: format!("directory listing for {path} could not be used: {detail}"),
1273 })?;
1274 if entries.is_empty() {
1277 return Err(SailError::Execution {
1278 code: RpcStatus::FailedPrecondition,
1279 detail: format!(
1280 "directory listing for {path} produced no records; \
1281 listing requires GNU find in the guest"
1282 ),
1283 });
1284 }
1285 let start = entries.remove(0);
1286 if start.entry_type != EntryType::Directory {
1287 return Err(SailError::Execution {
1288 code: RpcStatus::FailedPrecondition,
1289 detail: format!(
1290 "{path} is not a directory (it is a {})",
1291 start.entry_type.as_str()
1292 ),
1293 });
1294 }
1295 Ok(entries)
1296 }
1297}
1298
1299pub(crate) fn duration_to_whole_seconds(duration: Duration) -> i64 {
1303 duration.as_secs_f64().ceil() as i64
1304}
1305
1306fn fs_command_ok(result: &ExecResult, action: &str) -> Result<(), SailError> {
1308 if result.exit_code != 0 {
1309 return Err(fs_command_error(result, action));
1310 }
1311 Ok(())
1312}
1313
1314fn fs_command_error(result: &ExecResult, action: &str) -> SailError {
1317 let stderr = result.stderr.trim();
1318 let suffix = if stderr.is_empty() {
1319 String::new()
1320 } else {
1321 format!(": {stderr}")
1322 };
1323 SailError::Execution {
1324 code: RpcStatus::FailedPrecondition,
1325 detail: format!(
1326 "failed to {action} (exit code {}){suffix}",
1327 result.exit_code
1328 ),
1329 }
1330}
1331
1332fn should_reresolve_hinted_exec_endpoint(err: &SailError) -> bool {
1336 err.retryable()
1337 || matches!(
1338 err,
1339 SailError::Terminated { .. } | SailError::HostLost { .. }
1340 )
1341 || matches!(
1342 err,
1343 SailError::Execution {
1344 code: RpcStatus::Unknown | RpcStatus::Internal,
1345 detail,
1346 } if is_transient_transport_message(detail)
1347 )
1348}
1349
1350#[cfg(test)]
1351mod fs_exec_options_tests {
1352 use super::*;
1353
1354 #[test]
1355 fn no_user_pins_kernel_root() {
1356 let options = Client::fs_exec_options(None);
1357 assert_eq!(options.user.as_deref(), Some("0:0"));
1358 }
1359
1360 #[test]
1361 fn caller_user_runs_the_helper_as_that_user() {
1362 let options = Client::fs_exec_options(Some("alice:staff".to_string()));
1363 assert_eq!(options.user.as_deref(), Some("alice:staff"));
1364 }
1365
1366 #[test]
1367 fn empty_user_counts_as_unspecified_and_pins_kernel_root() {
1368 let options = Client::fs_exec_options(Some(String::new()));
1371 assert_eq!(options.user.as_deref(), Some("0:0"));
1372 }
1373}
1374
1375#[cfg(test)]
1376mod timeout_tests {
1377 use super::*;
1378
1379 #[test]
1380 fn durations_round_up_to_whole_seconds() {
1381 assert_eq!(duration_to_whole_seconds(Duration::ZERO), 0);
1382 assert_eq!(duration_to_whole_seconds(Duration::from_millis(500)), 1);
1383 assert_eq!(duration_to_whole_seconds(Duration::from_millis(1500)), 2);
1384 assert_eq!(duration_to_whole_seconds(Duration::from_mins(1)), 60);
1385 }
1386
1387 #[test]
1388 fn hinted_exec_reresolves_source_less_transport_statuses() {
1389 let relayed_transport = SailError::Execution {
1390 code: RpcStatus::Unknown,
1391 detail: "error reading server preface: EOF".to_string(),
1392 };
1393 assert!(should_reresolve_hinted_exec_endpoint(&relayed_transport));
1394
1395 let server_verdict = SailError::Execution {
1396 code: RpcStatus::Unknown,
1397 detail: "application rejected exec".to_string(),
1398 };
1399 assert!(!should_reresolve_hinted_exec_endpoint(&server_verdict));
1400 }
1401}