use super::*;
impl VmLocalExecutionBackend {
pub(super) async fn inspect_sandbox(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<LocalExecutionObservation> {
self.metadata(record)?;
if let Some(manager) = self.manager(&record.id) {
return self.inspect_registered(record, manager).await;
}
let home_dir = self.home_dir.clone();
let box_dir = record.box_dir.clone();
let box_id = record.id.clone();
let execution_id = execution_id(record)?;
let state = tokio::task::spawn_blocking(move || {
inspect_recorded_sandbox(&home_dir, &box_dir, &box_id)
})
.await
.map_err(|error| {
ExecutionManagerError::Internal(format!(
"Sandbox inspection task failed for {}: {error}",
record.id
))
})?
.map_err(|error| runtime_error("inspect", record, error))?
.ok_or(ExecutionManagerError::NotFound(execution_id))?;
match state.status.as_str() {
"created" | "running" => {
if state.pid == 0 {
return Err(ExecutionManagerError::Internal(format!(
"Sandbox runtime returned PID zero for {}",
record.id
)));
}
let manager = self.attach_sandbox(record, state).await?;
self.inspect_registered(record, manager).await
}
"paused" => {
if state.pid == 0 {
return Err(ExecutionManagerError::Internal(format!(
"Sandbox runtime returned PID zero for {}",
record.id
)));
}
let manager = self.attach_sandbox(record, state).await?;
let manager = manager.lock().await;
let handle = self.handle_from_manager(record, &manager).await?;
Ok(LocalExecutionObservation {
state: ExecutionState::Paused,
handle: Some(handle),
exit_code: None,
})
}
"stopped" => {
let exit_code = match crate::rootfs::read_persisted_exit_code(&record.box_dir) {
Some(exit_code) => exit_code,
None => {
self.collect_detached_sandbox_exit_code(record, &state)
.await?
}
};
self.cleanup_detached_sandbox(record).await?;
Ok(LocalExecutionObservation {
state: ExecutionState::Stopped,
handle: None,
exit_code: Some(exit_code),
})
}
status => Err(ExecutionManagerError::Internal(format!(
"Sandbox runtime returned unknown state {status} for {}",
record.id
))),
}
}
#[cfg(target_os = "linux")]
pub(super) async fn pause_sandbox(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<LocalExecutionHandle> {
self.transition_sandbox(record, true).await
}
#[cfg(target_os = "linux")]
pub(super) async fn resume_sandbox(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<LocalExecutionHandle> {
self.transition_sandbox(record, false).await
}
#[cfg(not(target_os = "linux"))]
pub(super) async fn pause_sandbox(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<LocalExecutionHandle> {
Err(unsupported(
record,
"pause",
"the Sandbox backend on this host",
))
}
#[cfg(not(target_os = "linux"))]
pub(super) async fn resume_sandbox(
&self,
record: &BoxRecord,
) -> ExecutionManagerResult<LocalExecutionHandle> {
Err(unsupported(
record,
"resume",
"the Sandbox backend on this host",
))
}
#[cfg(target_os = "linux")]
async fn transition_sandbox(
&self,
record: &BoxRecord,
pause: bool,
) -> ExecutionManagerResult<LocalExecutionHandle> {
self.metadata(record)?;
let home_dir = self.home_dir.clone();
let box_dir = record.box_dir.clone();
let box_id = record.id.clone();
let operation = if pause { "pause" } else { "resume" };
let inspection = tokio::task::spawn_blocking(move || {
let inspection =
inspect_recorded_sandbox(&home_dir, &box_dir, &box_id)?.ok_or_else(|| {
a3s_box_core::BoxError::StateError(format!(
"Sandbox runtime record is missing for {box_id}"
))
})?;
let socket = inspection.runtime.runtime_socket.as_ref().ok_or_else(|| {
a3s_box_core::BoxError::StateError(format!(
"A3S OCI runtime socket is missing for {box_id}"
))
})?;
let generation = inspection.runtime.generation.ok_or_else(|| {
a3s_box_core::BoxError::StateError(format!(
"A3S OCI generation is missing for {box_id}"
))
})?;
if pause {
crate::sandbox::a3s_oci_handler::A3sOciHandler::pause_at(
socket, &box_id, generation,
)?;
} else {
crate::sandbox::a3s_oci_handler::A3sOciHandler::resume_at(
socket, &box_id, generation,
)?;
}
inspect_recorded_sandbox(&home_dir, &box_dir, &box_id)?.ok_or_else(|| {
a3s_box_core::BoxError::StateError(format!(
"Sandbox runtime record disappeared after {operation} for {box_id}"
))
})
})
.await
.map_err(|error| {
ExecutionManagerError::Internal(format!(
"Sandbox {operation} task failed for {}: {error}",
record.id
))
})?
.map_err(|error| runtime_error(operation, record, error))?;
let expected = if pause { "paused" } else { "running" };
if inspection.status != expected {
return Err(ExecutionManagerError::Internal(format!(
"Sandbox runtime returned state {} after {operation} for {}",
inspection.status, record.id
)));
}
let manager = self.attach_sandbox(record, inspection).await?;
let manager = manager.lock().await;
self.handle_from_manager(record, &manager).await
}
#[cfg(target_os = "linux")]
async fn attach_sandbox(
&self,
record: &BoxRecord,
inspection: SandboxInspection,
) -> ExecutionManagerResult<SharedVm> {
let mut manager = self.new_manager(record)?;
let socket_dir = crate::vm::runtime_socket_dir(&self.home_dir, &record.id);
manager.exec_socket_path = Some(socket_dir.join("exec.sock"));
manager.pty_socket_path = Some(socket_dir.join("pty.sock"));
manager.port_forward_socket_path = Some(socket_dir.join("portfwd.sock"));
let runtime_socket = inspection.runtime.runtime_socket.ok_or_else(|| {
ExecutionManagerError::Internal(format!(
"A3S OCI runtime socket is missing for {}",
record.id
))
})?;
let generation = inspection.runtime.generation.ok_or_else(|| {
ExecutionManagerError::Internal(format!(
"A3S OCI generation is missing for {}",
record.id
))
})?;
let owner_pid = inspection.runtime.owner_pid.ok_or_else(|| {
ExecutionManagerError::Internal(format!(
"A3S OCI owner PID is missing for {}",
record.id
))
})?;
let owner_pid_start_time = inspection.runtime.owner_pid_start_time.ok_or_else(|| {
ExecutionManagerError::Internal(format!(
"A3S OCI owner identity is missing for {}",
record.id
))
})?;
let container_id = a3s_oci_sdk::ContainerId::new(record.id.clone())
.map_err(|error| ExecutionManagerError::Internal(error.to_string()))?;
let handler: Box<dyn a3s_box_core::vmm::VmHandler> = Box::new(
crate::sandbox::a3s_oci_handler::A3sOciHandler::from_recorded_runtime(
crate::sandbox::a3s_oci_handler::A3sOciHandlerSpec {
runtime_socket,
runtime_root: inspection.runtime.runtime_root,
container_id,
generation: a3s_oci_sdk::Generation(generation),
init_pid: inspection.pid,
owner_pid,
owner_pid_start_time,
bundle_dir: inspection.runtime.bundle_dir,
runtime_record: record.box_dir.join("sandbox/runtime.json"),
},
inspection.runtime.log_worker_pid,
inspection.runtime.log_worker_pid_start_time,
)
.await
.map_err(|error| runtime_error("recover", record, error))?,
);
*manager.handler.write().await = Some(handler);
if !matches!(
managed_state(record)?,
ManagedExecutionState::Starting | ManagedExecutionState::RestartStarting
) {
*manager.state.write().await = crate::BoxState::Ready;
}
let recovered = Arc::new(Mutex::new(manager));
match self.managers.entry(record.id.clone()) {
Entry::Occupied(entry) => Ok(Arc::clone(entry.get())),
Entry::Vacant(entry) => {
entry.insert(Arc::clone(&recovered));
Ok(recovered)
}
}
}
#[cfg(not(target_os = "linux"))]
async fn attach_sandbox(
&self,
record: &BoxRecord,
_inspection: SandboxInspection,
) -> ExecutionManagerResult<SharedVm> {
Err(unsupported(
record,
"recovery",
"the Sandbox backend on this host",
))
}
pub(super) async fn destroy_detached_sandbox(
&self,
record: &BoxRecord,
remove_anonymous_volumes: bool,
force_preserve_rootfs: bool,
timeout_secs: Option<u64>,
) -> ExecutionManagerResult<LocalExecutionTermination> {
let observation = self.inspect_sandbox(record).await?;
if let Some(manager) = self.manager(&record.id) {
return self
.destroy_registered(
record,
manager,
remove_anonymous_volumes,
force_preserve_rootfs,
timeout_secs,
)
.await;
}
if remove_anonymous_volumes {
let anonymous_volumes = self.anonymous_volumes_for_record(record).await;
self.cleanup_anonymous_volumes(anonymous_volumes).await;
}
Ok(LocalExecutionTermination {
outcome: KillOutcome::Killed,
exit_code: observation.exit_code,
})
}
#[cfg(target_os = "linux")]
async fn collect_detached_sandbox_exit_code(
&self,
record: &BoxRecord,
state: &SandboxInspection,
) -> ExecutionManagerResult<i32> {
let runtime_socket = state.runtime.runtime_socket.clone().ok_or_else(|| {
ExecutionManagerError::Internal(format!(
"A3S OCI runtime socket is missing for {}",
record.id
))
})?;
let generation = state.runtime.generation.ok_or_else(|| {
ExecutionManagerError::Internal(format!(
"A3S OCI generation is missing for {}",
record.id
))
})?;
let box_id = record.id.clone();
let exit_code =
tokio::task::spawn_blocking(move || -> a3s_box_core::Result<Option<i32>> {
let deadline = std::time::Instant::now() + TERMINAL_EXIT_POLL_TIMEOUT;
loop {
if let Some(exit_code) = crate::sandbox::A3sOciHandler::try_wait_at(
&runtime_socket,
&box_id,
generation,
)? {
return Ok(Some(exit_code));
}
if std::time::Instant::now() >= deadline {
return Ok(None);
}
std::thread::sleep(TERMINAL_EXIT_POLL_INTERVAL);
}
})
.await
.map_err(|error| {
ExecutionManagerError::Unavailable(format!(
"Sandbox exit-status task failed for {}: {error}",
record.id
))
})?
.map_err(|error| {
ExecutionManagerError::Unavailable(format!(
"failed to collect exact Sandbox exit status for {}: {error}",
record.id
))
})?;
exit_code.ok_or_else(|| {
ExecutionManagerError::Unavailable(format!(
"Sandbox reported execution {} as terminal before its exact exit status became available",
record.id
))
})
}
#[cfg(not(target_os = "linux"))]
async fn collect_detached_sandbox_exit_code(
&self,
record: &BoxRecord,
_state: &SandboxInspection,
) -> ExecutionManagerResult<i32> {
Err(unsupported(
record,
"exit-status recovery",
"the Sandbox backend on this host",
))
}
async fn cleanup_detached_sandbox(&self, record: &BoxRecord) -> ExecutionManagerResult<()> {
let home_dir = self.home_dir.clone();
let box_dir = record.box_dir.clone();
let box_id = record.id.clone();
tokio::task::spawn_blocking(move || {
crate::vm::reap::cleanup_recorded_sandbox_runtime_in(&home_dir, &box_dir, &box_id)
})
.await
.map_err(|error| {
ExecutionManagerError::Internal(format!(
"Sandbox cleanup task failed for {}: {error}",
record.id
))
})?
.map_err(|error| runtime_error("kill", record, error))?;
let mut manager = self.new_manager(record)?;
if should_force_rootfs_preservation(record)? {
manager
.destroy_preserving_rootfs()
.await
.map_err(|error| runtime_error("clean up", record, error))?;
} else {
manager
.destroy()
.await
.map_err(|error| runtime_error("clean up", record, error))?;
}
Ok(())
}
}
#[cfg(target_os = "linux")]
struct SandboxInspection {
status: String,
pid: u32,
runtime: crate::vm::reap::RecordedSandboxRuntime,
}
#[cfg(target_os = "linux")]
fn inspect_recorded_sandbox(
home_dir: &Path,
box_dir: &Path,
box_id: &str,
) -> a3s_box_core::Result<Option<SandboxInspection>> {
let Some(runtime) = crate::vm::reap::load_recorded_sandbox_runtime(home_dir, box_dir, box_id)?
else {
return Ok(None);
};
let socket = runtime.runtime_socket.as_ref().ok_or_else(|| {
a3s_box_core::BoxError::StateError(format!(
"A3S OCI runtime socket is missing for {box_id}"
))
})?;
let generation = runtime.generation.ok_or_else(|| {
a3s_box_core::BoxError::StateError(format!("A3S OCI generation is missing for {box_id}"))
})?;
let (status, pid) = match crate::sandbox::a3s_oci_handler::A3sOciHandler::query_state_at(
socket, box_id, generation,
)? {
Some(state) => (state.status, state.pid),
None => ("stopped".to_string(), 0),
};
if matches!(status.as_str(), "created" | "running" | "paused") && pid != runtime.init_pid {
return Err(a3s_box_core::BoxError::StateError(format!(
"Sandbox runtime PID disagrees with its durable record for {box_id}"
)));
}
Ok(Some(SandboxInspection {
status,
pid,
runtime,
}))
}
#[cfg(not(target_os = "linux"))]
struct SandboxInspection {
status: String,
pid: u32,
}
#[cfg(not(target_os = "linux"))]
fn inspect_recorded_sandbox(
_home_dir: &Path,
_box_dir: &Path,
_box_id: &str,
) -> a3s_box_core::Result<Option<SandboxInspection>> {
Err(a3s_box_core::BoxError::StateError(
"Sandbox execution requires Linux".to_string(),
))
}