1#[path = "vm_sandbox.rs"]
4mod sandbox;
5
6use std::path::{Path, PathBuf};
7use std::sync::Arc;
8#[cfg(unix)]
9use std::time::Duration;
10
11use a3s_box_core::{
12 EventEmitter, ExecutionBackend, ExecutionId, ExecutionManagerError, ExecutionManagerResult,
13 ExecutionState, KillOutcome, DEFAULT_SHUTDOWN_TIMEOUT_MS,
14};
15use async_trait::async_trait;
16use dashmap::mapref::entry::Entry;
17use dashmap::DashMap;
18use tokio::sync::Mutex;
19
20use super::resources::ExecutionResourceGuard;
21use super::vm_process::{locate_microvm_process, LocatedProcess};
22use super::{LocalExecutionBackend, LocalExecutionHandle, LocalExecutionObservation};
23use crate::{
24 BoxRecord, ManagedExecutionMetadata, ManagedExecutionOperation, ManagedExecutionState,
25 VmManager,
26};
27
28type SharedVm = Arc<Mutex<VmManager>>;
29
30#[derive(Clone)]
33pub struct VmLocalExecutionBackend {
34 home_dir: PathBuf,
35 managers: Arc<DashMap<String, SharedVm>>,
36 pull_progress_fn: Option<crate::PullProgressFn>,
37}
38
39impl VmLocalExecutionBackend {
40 pub fn new(home_dir: impl Into<PathBuf>) -> Self {
41 Self {
42 home_dir: home_dir.into(),
43 managers: Arc::new(DashMap::new()),
44 pull_progress_fn: None,
45 }
46 }
47
48 pub fn with_pull_progress_fn(mut self, pull_progress_fn: crate::PullProgressFn) -> Self {
49 self.pull_progress_fn = Some(pull_progress_fn);
50 self
51 }
52
53 pub fn home_dir(&self) -> &Path {
54 &self.home_dir
55 }
56
57 fn metadata<'a>(
58 &self,
59 record: &'a BoxRecord,
60 ) -> ExecutionManagerResult<&'a ManagedExecutionMetadata> {
61 uuid::Uuid::parse_str(&record.id).map_err(|error| {
62 ExecutionManagerError::Internal(format!(
63 "managed execution has an invalid internal ID {}: {error}",
64 record.id
65 ))
66 })?;
67 let expected_box_dir = self.home_dir.join("boxes").join(&record.id);
68 if record.box_dir != expected_box_dir {
69 return Err(ExecutionManagerError::Internal(format!(
70 "managed execution {} has an unexpected host directory {}",
71 record.id,
72 record.box_dir.display()
73 )));
74 }
75 let metadata = record.managed_execution.as_ref().ok_or_else(|| {
76 ExecutionManagerError::Internal(format!(
77 "execution {} lost managed lifecycle metadata",
78 record.id
79 ))
80 })?;
81 metadata
82 .validate()
83 .map_err(|error| ExecutionManagerError::Internal(error.to_string()))?;
84 if record.isolation != metadata.request.config.isolation {
85 return Err(ExecutionManagerError::Internal(format!(
86 "managed execution {} has inconsistent isolation metadata",
87 record.id
88 )));
89 }
90 Ok(metadata)
91 }
92
93 fn new_manager(&self, record: &BoxRecord) -> ExecutionManagerResult<VmManager> {
94 let metadata = self.metadata(record)?;
95 let mut config = metadata.request.config.clone();
96 if let Some(shm_size) = metadata.request.policy.shm_size {
97 let has_shared_memory_mount = config
98 .tmpfs
99 .iter()
100 .any(|entry| entry.split(':').next() == Some("/dev/shm"));
101 if !has_shared_memory_mount {
102 config.tmpfs.push(format!("/dev/shm:size={shm_size}"));
103 }
104 }
105 let mut manager = VmManager::with_box_id(config, EventEmitter::new(256), record.id.clone());
106 manager.home_dir = self.home_dir.clone();
107 if let Some(pull_progress_fn) = self.pull_progress_fn.clone() {
108 manager.set_pull_progress_fn(pull_progress_fn);
109 }
110 manager.anonymous_volumes = record.anonymous_volumes.clone();
111 manager.set_log_config(record.log_config.clone());
112 manager.resolved_execution_plan = Some(metadata.plan.clone());
113 Ok(manager)
114 }
115
116 fn manager(&self, execution_id: &str) -> Option<SharedVm> {
117 self.managers
118 .get(execution_id)
119 .map(|entry| Arc::clone(entry.value()))
120 }
121
122 fn remove_manager(&self, execution_id: &str, expected: &SharedVm) {
123 if let Entry::Occupied(entry) = self.managers.entry(execution_id.to_string()) {
124 if Arc::ptr_eq(entry.get(), expected) {
125 entry.remove();
126 }
127 }
128 }
129
130 async fn handle_from_manager(
131 &self,
132 record: &BoxRecord,
133 manager: &VmManager,
134 ) -> ExecutionManagerResult<LocalExecutionHandle> {
135 let execution_id = execution_id(record)?;
136 let pid = manager.pid().await.ok_or_else(|| {
137 ExecutionManagerError::Internal(format!(
138 "runtime returned no host PID for {execution_id}"
139 ))
140 })?;
141 let pid_start_time = crate::process::pid_start_time(pid);
142 #[cfg(target_os = "linux")]
143 if pid_start_time.is_none() {
144 return Err(ExecutionManagerError::NotFound(execution_id));
145 }
146 if !crate::process::is_process_alive_with_identity(pid, pid_start_time) {
147 return Err(ExecutionManagerError::NotFound(execution_id));
148 }
149 let exec_socket_path = manager
150 .exec_socket_path()
151 .map(Path::to_path_buf)
152 .ok_or_else(|| {
153 ExecutionManagerError::Internal(format!(
154 "runtime returned no exec socket for {}",
155 record.id
156 ))
157 })?;
158 let anonymous_volumes = if manager.anonymous_volumes().is_empty() {
159 self.anonymous_volumes_for_record(record).await
160 } else {
161 manager.anonymous_volumes().to_vec()
162 };
163 Ok(LocalExecutionHandle {
164 started_at: record.started_at.unwrap_or_else(chrono::Utc::now),
165 pid: Some(pid),
166 pid_start_time,
167 exec_socket_path,
168 console_log: record.box_dir.join("logs/console.log"),
169 anonymous_volumes,
170 })
171 }
172
173 async fn inspect_registered(
174 &self,
175 record: &BoxRecord,
176 shared: SharedVm,
177 ) -> ExecutionManagerResult<LocalExecutionObservation> {
178 let mut manager = shared.lock().await;
179 let exit_code = manager
180 .try_wait_exit()
181 .await
182 .map_err(|error| runtime_error("inspect", record, error))?;
183 let mut state = manager.state().await;
184 let terminal = exit_code.is_some() || state == crate::BoxState::Stopped;
185 if terminal {
186 let cleanup = manager.destroy().await;
187 let exit_code = manager.exit_code().or(exit_code);
188 drop(manager);
189 self.remove_manager(&record.id, &shared);
190 cleanup.map_err(|error| runtime_error("clean up", record, error))?;
191 return Ok(LocalExecutionObservation {
192 state: ExecutionState::Stopped,
193 handle: None,
194 exit_code,
195 });
196 }
197
198 if state == crate::BoxState::Created {
199 if manager.has_exited().await {
200 let cleanup = manager.destroy().await;
201 let exit_code = manager.exit_code();
202 drop(manager);
203 self.remove_manager(&record.id, &shared);
204 cleanup.map_err(|error| runtime_error("clean up", record, error))?;
205 return Ok(LocalExecutionObservation {
206 state: ExecutionState::Stopped,
207 handle: None,
208 exit_code,
209 });
210 }
211 if !self.promote_if_ready(record, &mut manager).await {
212 return Ok(LocalExecutionObservation {
213 state: ExecutionState::Creating,
214 handle: None,
215 exit_code: None,
216 });
217 }
218 state = manager.state().await;
219 }
220
221 if !manager
222 .health_check()
223 .await
224 .map_err(|error| runtime_error("inspect", record, error))?
225 {
226 let cleanup = manager.destroy().await;
227 let exit_code = manager.exit_code();
228 drop(manager);
229 self.remove_manager(&record.id, &shared);
230 cleanup.map_err(|error| runtime_error("clean up", record, error))?;
231 return Ok(LocalExecutionObservation {
232 state: ExecutionState::Stopped,
233 handle: None,
234 exit_code,
235 });
236 }
237
238 if state != crate::BoxState::Ready
239 && state != crate::BoxState::Busy
240 && state != crate::BoxState::Compacting
241 {
242 return Err(ExecutionManagerError::Internal(format!(
243 "runtime manager for {} is in unexpected state {state:?}",
244 record.id
245 )));
246 }
247 if matches!(
248 managed_state(record)?,
249 ManagedExecutionState::Starting | ManagedExecutionState::RestartStarting
250 ) && !exec_endpoint_ready(manager.exec_socket_path()).await
251 {
252 return Ok(LocalExecutionObservation {
253 state: ExecutionState::Creating,
254 handle: None,
255 exit_code: None,
256 });
257 }
258 let visible_state = visible_active_state(record)?;
259 let handle = self.handle_from_manager(record, &manager).await?;
260 Ok(LocalExecutionObservation {
261 state: visible_state,
262 handle: Some(handle),
263 exit_code: None,
264 })
265 }
266
267 async fn promote_if_ready(&self, record: &BoxRecord, manager: &mut VmManager) -> bool {
268 let socket_dir = crate::vm::runtime_socket_dir(&self.home_dir, &record.id);
269 let exec_socket = socket_dir.join("exec.sock");
270 if !exec_endpoint_ready(Some(&exec_socket)).await {
271 return false;
272 }
273 manager.exec_socket_path = Some(exec_socket);
274 manager.pty_socket_path = Some(socket_dir.join("pty.sock"));
275 manager.port_forward_socket_path = Some(socket_dir.join("portfwd.sock"));
276 *manager.state.write().await = crate::BoxState::Ready;
277 true
278 }
279
280 async fn recover_microvm(&self, record: &BoxRecord) -> ExecutionManagerResult<SharedVm> {
281 self.metadata(record)?;
282 let execution_id = execution_id(record)?;
283 let execution_id_label = record.id.clone();
284 let recorded = record.pid.map(|pid| (pid, record.pid_start_time));
285 let located = tokio::task::spawn_blocking(move || {
286 locate_microvm_process(&execution_id_label, recorded)
287 })
288 .await
289 .map_err(|error| {
290 ExecutionManagerError::Internal(format!(
291 "MicroVM process discovery task failed for {}: {error}",
292 record.id
293 ))
294 })?
295 .map_err(ExecutionManagerError::Internal)?
296 .ok_or(ExecutionManagerError::NotFound(execution_id))?;
297 self.attach_microvm(record, located).await
298 }
299
300 async fn attach_microvm(
301 &self,
302 record: &BoxRecord,
303 located: LocatedProcess,
304 ) -> ExecutionManagerResult<SharedVm> {
305 let mut manager = self.new_manager(record)?;
306 let socket_dir = crate::vm::runtime_socket_dir(&self.home_dir, &record.id);
307 manager
308 .attach_running_process(
309 located.pid,
310 socket_dir.join("exec.sock"),
311 Some(socket_dir.join("pty.sock")),
312 )
313 .await
314 .map_err(|error| runtime_error("recover", record, error))?;
315 if located.start_time.is_some()
316 && crate::process::pid_start_time(located.pid) != located.start_time
317 {
318 return Err(ExecutionManagerError::NotFound(execution_id(record)?));
319 }
320 let recovered = Arc::new(Mutex::new(manager));
321 match self.managers.entry(record.id.clone()) {
322 Entry::Occupied(entry) => Ok(Arc::clone(entry.get())),
323 Entry::Vacant(entry) => {
324 entry.insert(Arc::clone(&recovered));
325 Ok(recovered)
326 }
327 }
328 }
329
330 async fn require_microvm(&self, record: &BoxRecord) -> ExecutionManagerResult<SharedVm> {
331 match self.manager(&record.id) {
332 Some(manager) => Ok(manager),
333 None => self.recover_microvm(record).await,
334 }
335 }
336
337 async fn destroy_registered(
338 &self,
339 record: &BoxRecord,
340 shared: SharedVm,
341 remove_anonymous_volumes: bool,
342 timeout_secs: Option<u64>,
343 ) -> ExecutionManagerResult<KillOutcome> {
344 let mut manager = shared.lock().await;
345 let mut anonymous_volumes = if manager.anonymous_volumes().is_empty() {
346 record.anonymous_volumes.clone()
347 } else {
348 manager.anonymous_volumes().to_vec()
349 };
350 let result = match graceful_stop_options(record, timeout_secs)? {
351 Some((signal, timeout_ms)) => manager.destroy_with_options(signal, timeout_ms).await,
352 None => manager.destroy().await,
353 };
354 drop(manager);
355 self.remove_manager(&record.id, &shared);
356 result.map_err(|error| runtime_error("kill", record, error))?;
357 if remove_anonymous_volumes {
358 if anonymous_volumes.is_empty() {
359 anonymous_volumes = self.anonymous_volumes_for_record(record).await;
360 }
361 self.cleanup_anonymous_volumes(anonymous_volumes).await;
362 }
363 Ok(KillOutcome::Killed)
364 }
365
366 async fn anonymous_volumes_for_record(&self, record: &BoxRecord) -> Vec<String> {
367 if !record.anonymous_volumes.is_empty() {
368 return record.anonymous_volumes.clone();
369 }
370 let home_dir = self.home_dir.clone();
371 let execution_id = record.id.clone();
372 let short_id = record.id.chars().take(8).collect::<String>();
373 let result = tokio::task::spawn_blocking(move || -> a3s_box_core::Result<Vec<String>> {
374 let store =
375 crate::VolumeStore::new(home_dir.join("volumes.json"), home_dir.join("volumes"));
376 let prefix = format!("anon_{short_id}_");
377 let mut names = store
378 .load()?
379 .into_values()
380 .filter(|volume| {
381 volume
382 .labels
383 .get("anonymous")
384 .is_some_and(|value| value == "true")
385 && (volume.in_use_by.iter().any(|id| id == &execution_id)
386 || volume.name.starts_with(&prefix))
387 })
388 .map(|volume| volume.name)
389 .collect::<Vec<_>>();
390 names.sort();
391 Ok(names)
392 })
393 .await;
394 match result {
395 Ok(Ok(names)) => names,
396 Ok(Err(error)) => {
397 tracing::warn!(
398 execution_id = %record.id,
399 %error,
400 "Failed to load anonymous volumes during managed cleanup"
401 );
402 Vec::new()
403 }
404 Err(error) => {
405 tracing::warn!(
406 execution_id = %record.id,
407 %error,
408 "Anonymous volume recovery task failed"
409 );
410 Vec::new()
411 }
412 }
413 }
414
415 async fn cleanup_anonymous_volumes(&self, names: Vec<String>) {
416 if names.is_empty() {
417 return;
418 }
419 let home_dir = self.home_dir.clone();
420 let task = tokio::task::spawn_blocking(move || {
421 let store = crate::VolumeStore::new(
422 home_dir.join("volumes.json"),
423 home_dir.join("volumes"),
424 );
425 for name in names {
426 if let Err(error) = store.remove(&name, true) {
427 tracing::warn!(volume = %name, %error, "Failed to remove managed anonymous volume");
428 }
429 }
430 })
431 .await;
432 if let Err(error) = task {
433 tracing::warn!(%error, "Anonymous volume cleanup task failed");
434 }
435 }
436}
437
438#[async_trait]
439impl LocalExecutionBackend for VmLocalExecutionBackend {
440 async fn start(&self, record: &BoxRecord) -> ExecutionManagerResult<LocalExecutionHandle> {
441 self.metadata(record)?;
442 let manager = Arc::new(Mutex::new(self.new_manager(record)?));
443 match self.managers.entry(record.id.clone()) {
444 Entry::Occupied(_) => {
445 return Err(ExecutionManagerError::Unavailable(format!(
446 "execution {} already has an in-process runtime owner",
447 record.id
448 )))
449 }
450 Entry::Vacant(entry) => {
451 entry.insert(Arc::clone(&manager));
452 }
453 }
454
455 let mut guard = manager.lock().await;
456 let resource_home = self.home_dir.clone();
457 let resource_record = record.clone();
458 let resources = match tokio::task::spawn_blocking(move || {
459 ExecutionResourceGuard::prepare(&resource_home, &resource_record)
460 })
461 .await
462 {
463 Ok(Ok(resources)) => resources,
464 Ok(Err(error)) => {
465 drop(guard);
466 self.remove_manager(&record.id, &manager);
467 return Err(error);
468 }
469 Err(error) => {
470 drop(guard);
471 self.remove_manager(&record.id, &manager);
472 return Err(ExecutionManagerError::Internal(format!(
473 "managed resource preparation task failed for {}: {error}",
474 record.id
475 )));
476 }
477 };
478 if let Err(error) = guard.boot().await {
479 drop(guard);
480 self.remove_manager(&record.id, &manager);
481 let rollback = tokio::task::spawn_blocking(move || resources.rollback()).await;
482 if let Err(rollback_error) = rollback {
483 tracing::warn!(
484 execution_id = %record.id,
485 %rollback_error,
486 "Managed resource rollback task failed"
487 );
488 }
489 return Err(runtime_error("start", record, error));
490 }
491 resources.disarm();
492 self.handle_from_manager(record, &guard).await
493 }
494
495 async fn inspect(
496 &self,
497 record: &BoxRecord,
498 ) -> ExecutionManagerResult<LocalExecutionObservation> {
499 let metadata = self.metadata(record)?;
500 if metadata.plan.backend == ExecutionBackend::Crun {
501 return self.inspect_sandbox(record).await;
502 }
503 if let Some(manager) = self.manager(&record.id) {
504 return self.inspect_registered(record, manager).await;
505 }
506 let manager = self.recover_microvm(record).await?;
507 self.inspect_registered(record, manager).await
508 }
509
510 async fn pause(
511 &self,
512 record: &BoxRecord,
513 keep_memory: bool,
514 ) -> ExecutionManagerResult<LocalExecutionHandle> {
515 let metadata = self.metadata(record)?;
516 if metadata.plan.backend == ExecutionBackend::Crun {
517 if !keep_memory {
518 return Err(unsupported(
519 record,
520 "pause without memory retention",
521 "the Sandbox backend",
522 ));
523 }
524 return self.pause_sandbox(record).await;
525 }
526 if !keep_memory {
527 return Err(unsupported(
528 record,
529 "pause without memory retention",
530 "the local MicroVM backend",
531 ));
532 }
533 let shared = self.require_microvm(record).await?;
534 let manager = shared.lock().await;
535 require_recorded_pid(record, &manager).await?;
536 manager
537 .pause()
538 .await
539 .map_err(|error| runtime_error("pause", record, error))?;
540 self.handle_from_manager(record, &manager).await
541 }
542
543 async fn resume(&self, record: &BoxRecord) -> ExecutionManagerResult<LocalExecutionHandle> {
544 let metadata = self.metadata(record)?;
545 if metadata.plan.backend == ExecutionBackend::Crun {
546 return self.resume_sandbox(record).await;
547 }
548 let shared = self.require_microvm(record).await?;
549 let manager = shared.lock().await;
550 require_recorded_pid(record, &manager).await?;
551 manager
552 .resume()
553 .await
554 .map_err(|error| runtime_error("resume", record, error))?;
555 self.handle_from_manager(record, &manager).await
556 }
557
558 async fn kill(&self, record: &BoxRecord) -> ExecutionManagerResult<KillOutcome> {
559 let metadata = self.metadata(record)?;
560 let remove_anonymous_volumes = record.auto_remove;
561 let timeout_secs = record.stop_timeout;
562 if let Some(manager) = self.manager(&record.id) {
563 return self
564 .destroy_registered(record, manager, remove_anonymous_volumes, timeout_secs)
565 .await;
566 }
567 match metadata.plan.backend {
568 ExecutionBackend::Crun => {
569 self.destroy_detached_sandbox(record, remove_anonymous_volumes, timeout_secs)
570 .await
571 }
572 ExecutionBackend::Krun => {
573 let manager = self.recover_microvm(record).await?;
574 self.destroy_registered(record, manager, remove_anonymous_volumes, timeout_secs)
575 .await
576 }
577 }
578 }
579
580 async fn stop_for_restart(
581 &self,
582 record: &BoxRecord,
583 timeout_secs: Option<u64>,
584 ) -> ExecutionManagerResult<KillOutcome> {
585 let metadata = self.metadata(record)?;
586 let timeout_secs = timeout_secs.or(record.stop_timeout);
587 if let Some(manager) = self.manager(&record.id) {
588 return self
589 .destroy_registered(record, manager, false, timeout_secs)
590 .await;
591 }
592 match metadata.plan.backend {
593 ExecutionBackend::Crun => {
594 self.destroy_detached_sandbox(record, false, timeout_secs)
595 .await
596 }
597 ExecutionBackend::Krun => {
598 let manager = self.recover_microvm(record).await?;
599 self.destroy_registered(record, manager, false, timeout_secs)
600 .await
601 }
602 }
603 }
604}
605
606fn graceful_stop_options(
607 record: &BoxRecord,
608 timeout_secs: Option<u64>,
609) -> ExecutionManagerResult<Option<(i32, u64)>> {
610 if timeout_secs.is_none() && record.stop_signal.is_none() {
611 return Ok(None);
612 }
613 let timeout_ms = timeout_secs
614 .unwrap_or(DEFAULT_SHUTDOWN_TIMEOUT_MS / 1_000)
615 .checked_mul(1_000)
616 .ok_or_else(|| {
617 ExecutionManagerError::InvalidRequest(format!(
618 "stop timeout is too large for execution {}",
619 record.id
620 ))
621 })?;
622 let signal = record
623 .stop_signal
624 .as_deref()
625 .map(a3s_box_core::vmm::parse_signal_name)
626 .unwrap_or(libc::SIGTERM);
627 Ok(Some((signal, timeout_ms)))
628}
629
630async fn require_recorded_pid(
631 record: &BoxRecord,
632 manager: &VmManager,
633) -> ExecutionManagerResult<()> {
634 let execution_id = execution_id(record)?;
635 let pid = manager
636 .pid()
637 .await
638 .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
639 if record.pid != Some(pid)
640 || !crate::process::is_process_alive_with_identity(pid, record.pid_start_time)
641 {
642 return Err(ExecutionManagerError::NotFound(execution_id));
643 }
644 Ok(())
645}
646
647fn visible_active_state(record: &BoxRecord) -> ExecutionManagerResult<ExecutionState> {
648 match managed_state(record)? {
649 ManagedExecutionState::Paused | ManagedExecutionState::Resuming => {
650 Ok(ExecutionState::Paused)
651 }
652 ManagedExecutionState::Starting
653 | ManagedExecutionState::RestartStarting
654 | ManagedExecutionState::Running
655 | ManagedExecutionState::Pausing
656 | ManagedExecutionState::Killing => Ok(ExecutionState::Running),
657 ManagedExecutionState::Snapshotting => match record
658 .managed_execution
659 .as_ref()
660 .and_then(|metadata| metadata.pending_operation.as_ref())
661 {
662 Some(ManagedExecutionOperation::Snapshot {
663 source_state: ManagedExecutionState::Running,
664 ..
665 }) => Ok(ExecutionState::Running),
666 Some(ManagedExecutionOperation::Snapshot {
667 source_state: ManagedExecutionState::Paused,
668 ..
669 }) => Ok(ExecutionState::Paused),
670 _ => Err(ExecutionManagerError::Internal(format!(
671 "execution {} has invalid snapshot metadata",
672 record.id
673 ))),
674 },
675 ManagedExecutionState::RestartStopping => match record
676 .managed_execution
677 .as_ref()
678 .and_then(|metadata| metadata.pending_operation.as_ref())
679 {
680 Some(ManagedExecutionOperation::Restart {
681 source_state: ManagedExecutionState::Paused,
682 ..
683 }) => Ok(ExecutionState::Paused),
684 Some(ManagedExecutionOperation::Restart {
685 source_state: ManagedExecutionState::Running,
686 ..
687 }) => Ok(ExecutionState::Running),
688 _ => Err(ExecutionManagerError::Internal(format!(
689 "execution {} has invalid restart teardown metadata",
690 record.id
691 ))),
692 },
693 state => Err(ExecutionManagerError::Internal(format!(
694 "execution {} has no active runtime in managed state {state}",
695 record.id
696 ))),
697 }
698}
699
700fn managed_state(record: &BoxRecord) -> ExecutionManagerResult<ManagedExecutionState> {
701 record
702 .managed_state()
703 .map_err(|error| ExecutionManagerError::Internal(error.to_string()))?
704 .ok_or_else(|| {
705 ExecutionManagerError::Internal(format!("execution {} is not managed", record.id))
706 })
707}
708
709fn execution_id(record: &BoxRecord) -> ExecutionManagerResult<ExecutionId> {
710 ExecutionId::new(record.id.clone())
711 .map_err(|error| ExecutionManagerError::Internal(error.to_string()))
712}
713
714fn runtime_error(
715 action: &str,
716 record: &BoxRecord,
717 error: impl std::fmt::Display,
718) -> ExecutionManagerError {
719 ExecutionManagerError::Internal(format!(
720 "failed to {action} execution {}: {error}",
721 record.id
722 ))
723}
724
725fn unsupported(record: &BoxRecord, operation: &str, backend: &str) -> ExecutionManagerError {
726 match execution_id(record) {
727 Ok(execution_id) => ExecutionManagerError::Conflict {
728 execution_id,
729 message: format!("{operation} is not supported by {backend}"),
730 },
731 Err(error) => error,
732 }
733}
734
735#[cfg(unix)]
736async fn exec_endpoint_ready(path: Option<&Path>) -> bool {
737 let Some(path) = path else {
738 return false;
739 };
740 let attempt = async {
741 let client = crate::ExecClient::connect(path).await.ok()?;
742 client.heartbeat().await.ok().filter(|ready| *ready)
743 };
744 tokio::time::timeout(Duration::from_millis(500), attempt)
745 .await
746 .ok()
747 .flatten()
748 .is_some()
749}
750
751#[cfg(not(unix))]
752async fn exec_endpoint_ready(path: Option<&Path>) -> bool {
753 path.is_some()
754}
755
756#[cfg(test)]
757#[path = "vm_backend_tests.rs"]
758mod tests;