Skip to main content

a3s_box_runtime/local_execution/
vm_backend.rs

1//! Production local execution backend backed by [`crate::VmManager`].
2
3#[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/// Runtime adapter that owns live [`VmManager`] handles and reconstructs them
31/// from durable runtime evidence after a control-plane restart.
32#[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;