use std::sync::Arc;
use a3s_box_core::{
pty::PtyRequest, ExecOutput, ExecRequest, ExecutionEventBatch, ExecutionEventsRequest,
ExecutionIsolation, ExecutionManagerError, ExecutionManagerResult, ExecutionProcess,
ExecutionProcessInventory, ExecutionResourceUpdate, ExecutionStats, FileRequest, FileResponse,
FilesystemRequest, FilesystemResponse, KillOutcome, OperationId,
};
use async_trait::async_trait;
use super::{
LocalExecutionBackend, LocalExecutionHandle, LocalExecutionObservation,
LocalExecutionTermination,
};
use crate::{BoxRecord, ManagedRuntimeRoute};
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum OciMigrationPolicy {
#[default]
LegacyOnly,
SandboxViaOci,
AllViaOci,
}
impl OciMigrationPolicy {
const fn route(self, isolation: ExecutionIsolation) -> ManagedRuntimeRoute {
match (self, isolation) {
(Self::LegacyOnly, _) | (Self::SandboxViaOci, ExecutionIsolation::Microvm) => {
ManagedRuntimeRoute::BoxVm
}
(Self::SandboxViaOci, ExecutionIsolation::Sandbox) | (Self::AllViaOci, _) => {
ManagedRuntimeRoute::OciSdk
}
}
}
}
#[derive(Clone)]
pub struct LocalExecutionBackendRouter {
legacy: Arc<dyn LocalExecutionBackend>,
oci: Arc<dyn LocalExecutionBackend>,
policy: OciMigrationPolicy,
}
impl LocalExecutionBackendRouter {
pub fn new(
legacy: Arc<dyn LocalExecutionBackend>,
oci: Arc<dyn LocalExecutionBackend>,
policy: OciMigrationPolicy,
) -> Self {
Self {
legacy,
oci,
policy,
}
}
#[must_use]
pub const fn policy(&self) -> OciMigrationPolicy {
self.policy
}
fn route_for_record(&self, record: &BoxRecord) -> ExecutionManagerResult<ManagedRuntimeRoute> {
resolved_runtime_route(record)
}
fn backend_for_record(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<&Arc<dyn LocalExecutionBackend>> {
match self.route_for_record(record)? {
ManagedRuntimeRoute::BoxVm => Ok(&self.legacy),
ManagedRuntimeRoute::OciSdk => Ok(&self.oci),
ManagedRuntimeRoute::Unspecified => unreachable!("route inference is exhaustive"),
}
}
}
pub(super) fn resolved_runtime_route(
record: &BoxRecord,
) -> ExecutionManagerResult<ManagedRuntimeRoute> {
let metadata = record.managed_execution.as_ref().ok_or_else(|| {
ExecutionManagerError::Internal(format!(
"execution {} has no managed lifecycle metadata for backend routing",
record.id
))
})?;
metadata
.validate()
.map_err(|error| ExecutionManagerError::Internal(error.to_string()))?;
match metadata.runtime_route {
ManagedRuntimeRoute::BoxVm => Ok(ManagedRuntimeRoute::BoxVm),
ManagedRuntimeRoute::OciSdk => Ok(ManagedRuntimeRoute::OciSdk),
ManagedRuntimeRoute::Unspecified if metadata.oci_runtime.is_some() => {
Ok(ManagedRuntimeRoute::OciSdk)
}
ManagedRuntimeRoute::Unspecified if record.exec_socket_path.as_os_str().is_empty() => {
Ok(ManagedRuntimeRoute::OciSdk)
}
ManagedRuntimeRoute::Unspecified => Ok(ManagedRuntimeRoute::BoxVm),
}
}
#[async_trait]
impl LocalExecutionBackend for LocalExecutionBackendRouter {
fn route_for_create(&self, record: &BoxRecord) -> ExecutionManagerResult<ManagedRuntimeRoute> {
if record.managed_execution.is_none() {
return Err(ExecutionManagerError::Internal(format!(
"new execution {} has no managed lifecycle metadata for backend routing",
record.id
)));
}
Ok(self.policy.route(record.isolation))
}
async fn preflight(&self, record: &BoxRecord) -> ExecutionManagerResult<()> {
self.backend_for_record(record)?.preflight(record).await
}
async fn start(&self, record: &BoxRecord) -> ExecutionManagerResult<LocalExecutionHandle> {
self.backend_for_record(record)?.start(record).await
}
async fn inspect(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<LocalExecutionObservation> {
self.backend_for_record(record)?.inspect(record).await
}
async fn pause(
&self,
record: &BoxRecord,
keep_memory: bool,
) -> ExecutionManagerResult<LocalExecutionHandle> {
self.backend_for_record(record)?
.pause(record, keep_memory)
.await
}
async fn resume(&self, record: &BoxRecord) -> ExecutionManagerResult<LocalExecutionHandle> {
self.backend_for_record(record)?.resume(record).await
}
async fn preflight_resource_update(
&self,
record: &BoxRecord,
update: &ExecutionResourceUpdate,
) -> ExecutionManagerResult<()> {
self.backend_for_record(record)?
.preflight_resource_update(record, update)
.await
}
async fn update_resources(
&self,
record: &BoxRecord,
operation_id: &OperationId,
update: &ExecutionResourceUpdate,
) -> ExecutionManagerResult<()> {
self.backend_for_record(record)?
.update_resources(record, operation_id, update)
.await
}
async fn list_processes(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<ExecutionProcessInventory> {
self.backend_for_record(record)?
.list_processes(record)
.await
}
async fn stats(&self, record: &BoxRecord) -> ExecutionManagerResult<ExecutionStats> {
self.backend_for_record(record)?.stats(record).await
}
async fn events(
&self,
record: &BoxRecord,
request: ExecutionEventsRequest,
) -> ExecutionManagerResult<ExecutionEventBatch> {
self.backend_for_record(record)?
.events(record, request)
.await
}
async fn execute(
&self,
record: &BoxRecord,
request: ExecRequest,
) -> ExecutionManagerResult<ExecOutput> {
self.backend_for_record(record)?
.execute(record, request)
.await
}
async fn start_process(
&self,
record: &BoxRecord,
request: ExecRequest,
) -> ExecutionManagerResult<ExecutionProcess> {
self.backend_for_record(record)?
.start_process(record, request)
.await
}
async fn start_pty(
&self,
record: &BoxRecord,
request: PtyRequest,
) -> ExecutionManagerResult<ExecutionProcess> {
self.backend_for_record(record)?
.start_pty(record, request)
.await
}
async fn transfer_file(
&self,
record: &BoxRecord,
request: FileRequest,
) -> ExecutionManagerResult<FileResponse> {
self.backend_for_record(record)?
.transfer_file(record, request)
.await
}
async fn filesystem(
&self,
record: &BoxRecord,
request: FilesystemRequest,
) -> ExecutionManagerResult<FilesystemResponse> {
self.backend_for_record(record)?
.filesystem(record, request)
.await
}
async fn prepare_quiescent_rootfs(&self, record: &BoxRecord) -> ExecutionManagerResult<()> {
self.backend_for_record(record)?
.prepare_quiescent_rootfs(record)
.await
}
async fn cleanup_quiescent_rootfs(&self, record: &BoxRecord) -> ExecutionManagerResult<()> {
self.backend_for_record(record)?
.cleanup_quiescent_rootfs(record)
.await
}
async fn stop_for_restart(
&self,
record: &BoxRecord,
timeout_secs: Option<u64>,
) -> ExecutionManagerResult<KillOutcome> {
self.backend_for_record(record)?
.stop_for_restart(record, timeout_secs)
.await
}
async fn kill(&self, record: &BoxRecord) -> ExecutionManagerResult<KillOutcome> {
self.backend_for_record(record)?.kill(record).await
}
async fn kill_with_status(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<LocalExecutionTermination> {
self.backend_for_record(record)?
.kill_with_status(record)
.await
}
}