mod handlers;
mod types;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use axum::extract::{Request, State};
use axum::http::Method;
use axum::http::StatusCode;
use axum::http::header::AUTHORIZATION;
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post, put};
use axum::{Json, Router};
use clap::Args;
use futures::StreamExt;
use tokio::sync::RwLock;
use tower_http::catch_panic::CatchPanicLayer;
use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer};
use boxlite::runtime::options::{InboundNetworkConfig, NetworkMode, OutboundNetworkConfig};
use boxlite::{
BoxCommand, BoxInfo, BoxOptions, BoxStatus, BoxliteRuntime, ExecStdin, Execution, LiteBox,
NetworkSpec, RootfsSpec,
};
use crate::cli::GlobalFlags;
use crate::defaults::{LOCAL_SERVE_HOST, LOCAL_SERVE_PORT};
use self::types::{BoxResponse, CreateBoxRequest, ErrorBody, ErrorDetail, ExecRequest};
fn parse_api_key(raw: &str) -> Result<String, String> {
if raw.is_empty() {
Err("API key must not be empty".to_string())
} else {
Ok(raw.to_string())
}
}
#[derive(Args, Debug)]
pub struct ServeArgs {
#[arg(long, default_value_t = LOCAL_SERVE_PORT)]
pub port: u16,
#[arg(long, default_value_t = LOCAL_SERVE_HOST.to_string())]
pub host: String,
#[arg(
long,
env = "BOXLITE_SERVE_API_KEY",
hide_env_values = true,
value_parser = parse_api_key
)]
pub api_key: Option<String>,
}
struct AppState {
runtime: BoxliteRuntime,
boxes: RwLock<HashMap<String, Arc<LiteBox>>>,
executions: RwLock<HashMap<String, Arc<ActiveExecution>>>,
api_key: Option<String>,
lifecycle: RwLock<HashMap<String, LifecyclePolicy>>,
last_activity: RwLock<HashMap<String, Instant>>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(in crate::commands::serve) struct LifecyclePolicy {
pub auto_stop: u32,
pub auto_delete: u32,
}
impl LifecyclePolicy {
fn is_empty(&self) -> bool {
self.auto_stop == 0 && self.auto_delete == 0
}
}
impl AppState {
pub(in crate::commands::serve) async fn set_lifecycle(
&self,
box_id: &str,
policy: LifecyclePolicy,
) {
if policy.is_empty() {
return;
}
self.lifecycle
.write()
.await
.insert(box_id.to_string(), policy);
}
pub(in crate::commands::serve) async fn lifecycle_of(&self, box_id: &str) -> LifecyclePolicy {
self.lifecycle
.read()
.await
.get(box_id)
.copied()
.unwrap_or_default()
}
pub(in crate::commands::serve) async fn box_response(&self, info: &BoxInfo) -> BoxResponse {
let mut resp = box_info_to_response(info);
let policy = self.lifecycle_of(info.id.as_ref()).await;
resp.auto_stop = policy.auto_stop;
resp.auto_delete = policy.auto_delete;
resp
}
pub(in crate::commands::serve) async fn record_box_activity(&self, box_id: &str) {
self.last_activity
.write()
.await
.insert(box_id.to_string(), Instant::now());
}
pub(in crate::commands::serve) async fn forget_box(&self, box_id: &str) {
self.lifecycle.write().await.remove(box_id);
self.last_activity.write().await.remove(box_id);
}
pub(in crate::commands::serve) async fn retain_known_boxes(
&self,
live: &[(String, Option<String>)],
) {
self.last_activity
.write()
.await
.retain(|spelling, _| box_named_by(spelling, live).is_some());
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub(in crate::commands::serve) enum SessionKind {
Exec,
Main,
}
pub(in crate::commands::serve) const MAIN_SESSION_ID_HEADER: &str = "x-boxlite-execution-id";
pub(in crate::commands::serve) struct ActiveExecution {
box_id: String,
kind: SessionKind,
execution: Execution,
stdin: tokio::sync::Mutex<Option<ExecStdin>>,
stdout_bus: Arc<BacklogBroadcast>,
stderr_bus: Arc<BacklogBroadcast>,
attach: tokio::sync::Mutex<AttachState>,
done: std::sync::atomic::AtomicBool,
done_tx: tokio::sync::watch::Sender<bool>,
exit_code: std::sync::atomic::AtomicI32,
done_at: std::sync::Mutex<Option<Instant>>,
created_at: Instant,
}
struct AttachState {
connected: bool,
last_disconnect_at: Option<Instant>,
signaled_hup: bool,
signaled_term: bool,
reaping_kill: bool,
escalating: bool,
}
const ATTACH_BROADCAST_CAPACITY: usize = 256;
const BACKLOG_BYTE_CAP: usize = 256 * 1024;
struct BacklogBroadcast {
tx: tokio::sync::broadcast::Sender<Vec<u8>>,
state: std::sync::Mutex<BacklogState>,
cap: usize,
}
struct BacklogState {
backlog: std::collections::VecDeque<Vec<u8>>,
total_bytes: usize,
}
impl BacklogBroadcast {
fn new(capacity: usize, backlog_cap: usize) -> Self {
let (tx, _) = tokio::sync::broadcast::channel(capacity);
Self {
tx,
state: std::sync::Mutex::new(BacklogState {
backlog: std::collections::VecDeque::new(),
total_bytes: 0,
}),
cap: backlog_cap,
}
}
fn send(&self, data: Vec<u8>) {
let mut state = self.state.lock().unwrap();
state.total_bytes += data.len();
state.backlog.push_back(data.clone());
while state.total_bytes > self.cap && state.backlog.len() > 1 {
if let Some(old) = state.backlog.pop_front() {
state.total_bytes -= old.len();
} else {
break;
}
}
let _ = self.tx.send(data);
}
fn subscribe(&self) -> BacklogReceiver {
let state = self.state.lock().unwrap();
let replay: std::collections::VecDeque<Vec<u8>> = state.backlog.iter().cloned().collect();
let rx = self.tx.subscribe();
BacklogReceiver { replay, rx }
}
}
struct BacklogReceiver {
replay: std::collections::VecDeque<Vec<u8>>,
rx: tokio::sync::broadcast::Receiver<Vec<u8>>,
}
impl BacklogReceiver {
async fn recv(&mut self) -> Result<Vec<u8>, tokio::sync::broadcast::error::RecvError> {
if let Some(chunk) = self.replay.pop_front() {
return Ok(chunk);
}
self.rx.recv().await
}
fn try_recv(&mut self) -> Result<Vec<u8>, tokio::sync::broadcast::error::TryRecvError> {
if let Some(chunk) = self.replay.pop_front() {
return Ok(chunk);
}
self.rx.try_recv()
}
}
impl ActiveExecution {
fn new(
box_id: String,
kind: SessionKind,
mut execution: Execution,
stdin: Option<ExecStdin>,
) -> Arc<Self> {
let stdout = execution.stdout();
let stderr = execution.stderr();
let stdout_bus = Arc::new(BacklogBroadcast::new(
ATTACH_BROADCAST_CAPACITY,
BACKLOG_BYTE_CAP,
));
let stderr_bus = Arc::new(BacklogBroadcast::new(
ATTACH_BROADCAST_CAPACITY,
BACKLOG_BYTE_CAP,
));
let (done_tx, _) = tokio::sync::watch::channel(false);
let now = Instant::now();
let active = Arc::new(Self {
box_id,
kind,
execution,
stdin: tokio::sync::Mutex::new(stdin),
stdout_bus: stdout_bus.clone(),
stderr_bus: stderr_bus.clone(),
attach: tokio::sync::Mutex::new(AttachState {
connected: false,
last_disconnect_at: Some(now),
signaled_hup: false,
signaled_term: false,
reaping_kill: false,
escalating: false,
}),
done: std::sync::atomic::AtomicBool::new(false),
done_tx,
exit_code: std::sync::atomic::AtomicI32::new(-1),
done_at: std::sync::Mutex::new(None),
created_at: now,
});
let stdout_handle = if let Some(mut out) = stdout {
let bus = stdout_bus;
Some(tokio::spawn(async move {
while let Some(line) = out.next().await {
bus.send(line.into_bytes());
}
}))
} else {
None
};
let stderr_handle = if let Some(mut err) = stderr {
let bus = stderr_bus;
Some(tokio::spawn(async move {
while let Some(line) = err.next().await {
bus.send(line.into_bytes());
}
}))
} else {
None
};
{
let active = Arc::clone(&active);
tokio::spawn(async move {
if let Ok(result) = active.execution.wait().await {
active
.exit_code
.store(result.exit_code, std::sync::atomic::Ordering::SeqCst);
}
if let Some(h) = stdout_handle {
let _ = h.await;
}
if let Some(h) = stderr_handle {
let _ = h.await;
}
*active.done_at.lock().unwrap() = Some(Instant::now());
active.done.store(true, std::sync::atomic::Ordering::SeqCst);
let _ = active.done_tx.send(true);
});
}
active
}
pub(in crate::commands::serve) fn box_id(&self) -> &str {
&self.box_id
}
pub(in crate::commands::serve) fn kind(&self) -> SessionKind {
self.kind
}
pub(in crate::commands::serve) fn stdout_bus(&self) -> &BacklogBroadcast {
&self.stdout_bus
}
pub(in crate::commands::serve) fn stderr_bus(&self) -> &BacklogBroadcast {
&self.stderr_bus
}
pub(in crate::commands::serve) fn stdin(&self) -> &tokio::sync::Mutex<Option<ExecStdin>> {
&self.stdin
}
pub(in crate::commands::serve) fn execution(&self) -> &Execution {
&self.execution
}
pub(in crate::commands::serve) fn is_done(&self) -> bool {
self.done.load(std::sync::atomic::Ordering::SeqCst)
}
pub(in crate::commands::serve) fn done_rx(&self) -> tokio::sync::watch::Receiver<bool> {
self.done_tx.subscribe()
}
pub(in crate::commands::serve) fn exit_code(&self) -> i32 {
self.exit_code.load(std::sync::atomic::Ordering::SeqCst)
}
pub(in crate::commands::serve) fn created_at(&self) -> Instant {
self.created_at
}
pub(in crate::commands::serve) fn should_retain(&self, now: Instant) -> bool {
if !self.is_done() {
return true;
}
let done_at = self.done_at.lock().unwrap();
match *done_at {
Some(at) => now.duration_since(at) <= COMPLETED_RETENTION_GRACE,
None => true,
}
}
pub(in crate::commands::serve) async fn mark_connected(&self) -> bool {
let mut s = self.attach.lock().await;
if s.connected || s.reaping_kill || s.escalating {
return false;
}
s.connected = true;
s.last_disconnect_at = None;
s.signaled_hup = false;
s.signaled_term = false;
true
}
pub(in crate::commands::serve) async fn mark_disconnected(&self) {
let mut s = self.attach.lock().await;
s.connected = false;
s.last_disconnect_at = Some(Instant::now());
}
pub(in crate::commands::serve) async fn mark_reaping_kill(&self) {
let mut s = self.attach.lock().await;
s.reaping_kill = true;
}
async fn is_reaping_kill(&self) -> bool {
let s = self.attach.lock().await;
s.reaping_kill
}
async fn try_escalate_hup(&self, now: Instant, reconnect_grace: std::time::Duration) -> bool {
let mut s = self.attach.lock().await;
if s.connected || s.signaled_hup || s.reaping_kill || s.escalating {
return false;
}
let Some(disc) = s.last_disconnect_at else {
return false;
};
if now.duration_since(disc) <= reconnect_grace {
return false;
}
s.signaled_hup = true;
s.escalating = true;
s.last_disconnect_at = Some(now);
true
}
async fn try_escalate_term(&self, now: Instant, shutdown_grace: std::time::Duration) -> bool {
let mut s = self.attach.lock().await;
if s.connected || !s.signaled_hup || s.signaled_term || s.reaping_kill || s.escalating {
return false;
}
let Some(disc) = s.last_disconnect_at else {
return false;
};
if now.duration_since(disc) <= shutdown_grace {
return false;
}
s.signaled_term = true;
s.escalating = true;
s.last_disconnect_at = Some(now);
true
}
async fn try_escalate_kill(&self, now: Instant, shutdown_grace: std::time::Duration) -> bool {
let mut s = self.attach.lock().await;
if s.connected || !s.signaled_term || s.reaping_kill {
return false;
}
let Some(disc) = s.last_disconnect_at else {
return false;
};
if now.duration_since(disc) <= shutdown_grace {
return false;
}
s.reaping_kill = true;
true
}
async fn finish_escalation(&self) {
let mut s = self.attach.lock().await;
s.escalating = false;
}
async fn escalation_failed_mark_doomed(&self) {
let mut s = self.attach.lock().await;
s.escalating = false;
s.reaping_kill = true;
}
}
const REAPER_TICK: std::time::Duration = std::time::Duration::from_secs(30);
const REAPER_SIGNAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const LIFECYCLE_OP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
const DEFAULT_RECONNECT_GRACE: std::time::Duration = std::time::Duration::from_secs(300);
const DEFAULT_SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(30);
const DEFAULT_MAX_SESSION_LIFETIME: std::time::Duration = std::time::Duration::from_secs(24 * 3600);
const COMPLETED_RETENTION_GRACE: std::time::Duration = std::time::Duration::from_secs(5 * 60);
fn resolve_duration(var: &str, fallback: std::time::Duration) -> std::time::Duration {
let raw = match std::env::var(var) {
Ok(s) if !s.is_empty() => s,
_ => return fallback,
};
let parsed = if let Some(rest) = raw.strip_suffix('h') {
rest.parse::<u64>()
.ok()
.map(|n| std::time::Duration::from_secs(n * 3600))
} else if let Some(rest) = raw.strip_suffix('m') {
rest.parse::<u64>()
.ok()
.map(|n| std::time::Duration::from_secs(n * 60))
} else if let Some(rest) = raw.strip_suffix('s') {
rest.parse::<u64>().ok().map(std::time::Duration::from_secs)
} else {
raw.parse::<u64>().ok().map(std::time::Duration::from_secs)
};
match parsed {
Some(d) => d,
None => {
tracing::warn!(env = var, value = %raw,
"invalid duration env var (use Ns/Nm/Nh), using default");
fallback
}
}
}
#[derive(Debug, PartialEq, Eq)]
enum LifecycleAction {
Leave,
Stop,
Delete,
}
fn decide_lifecycle(
status: BoxStatus,
auto_stop_secs: u32,
auto_delete_secs: u32,
idle: std::time::Duration,
since_stop: std::time::Duration,
) -> LifecycleAction {
if status == BoxStatus::Running
&& auto_stop_secs > 0
&& idle >= std::time::Duration::from_secs(u64::from(auto_stop_secs))
{
return LifecycleAction::Stop;
}
let at_rest = matches!(status, BoxStatus::Stopped | BoxStatus::Failed);
if at_rest
&& auto_delete_secs > 0
&& since_stop >= std::time::Duration::from_secs(u64::from(auto_delete_secs))
{
return LifecycleAction::Delete;
}
LifecycleAction::Leave
}
async fn idle_or_seed(
last_activity: &RwLock<HashMap<String, Instant>>,
box_id: &str,
boxes: &[(String, Option<String>)],
now: Instant,
) -> std::time::Duration {
let stamped = {
let clocks = last_activity.read().await;
clocks
.iter()
.filter(|(spelling, _)| box_named_by(spelling, boxes) == Some(box_id))
.map(|(_, stamped)| *stamped)
.max()
};
match stamped {
Some(stamped) => now.saturating_duration_since(stamped),
None => {
last_activity.write().await.insert(box_id.to_string(), now);
std::time::Duration::ZERO
}
}
}
fn box_named_by<'a>(spelling: &str, boxes: &'a [(String, Option<String>)]) -> Option<&'a str> {
if spelling.is_empty() {
return None;
}
if let Some((id, _)) = boxes.iter().find(|(id, _)| id == spelling) {
return Some(id);
}
if let Some((id, _)) = boxes
.iter()
.find(|(_, name)| name.as_deref() == Some(spelling))
{
return Some(id);
}
let mut prefixed = boxes.iter().filter(|(id, _)| id.starts_with(spelling));
match (prefixed.next(), prefixed.next()) {
(Some((id, _)), None) => Some(id),
_ => None,
}
}
fn busy_box_ids(
executions: &HashMap<String, Arc<ActiveExecution>>,
) -> std::collections::HashSet<String> {
executions
.values()
.filter(|execution| execution.kind() != SessionKind::Main && !execution.is_done())
.map(|execution| execution.box_id.clone())
.collect()
}
async fn run_lifecycle_once(state: &AppState, now: Instant) {
let boxes = match tokio::time::timeout(LIFECYCLE_OP_TIMEOUT, state.runtime.list_info()).await {
Ok(Ok(boxes)) => boxes,
Ok(Err(error)) => {
tracing::warn!(%error, "lifecycle sweep could not list boxes");
return;
}
Err(_) => {
tracing::warn!("lifecycle sweep timed out listing boxes");
return;
}
};
let busy = busy_box_ids(&*state.executions.read().await);
let live: Vec<(String, Option<String>)> = boxes
.iter()
.map(|info| (info.id.to_string(), info.name.clone()))
.collect();
state.retain_known_boxes(&live).await;
let now_utc = chrono::Utc::now();
for info in boxes {
let box_id = info.id.to_string();
let policy = state.lifecycle_of(&box_id).await;
if policy.is_empty() {
continue;
}
if busy
.iter()
.any(|spelling| box_named_by(spelling, &live) == Some(box_id.as_str()))
{
continue;
}
let idle = idle_or_seed(&state.last_activity, &box_id, &live, now).await;
let since_stop = (now_utc - info.last_updated)
.to_std()
.unwrap_or(std::time::Duration::ZERO);
match decide_lifecycle(
info.status,
policy.auto_stop,
policy.auto_delete,
idle,
since_stop,
) {
LifecycleAction::Leave => {}
LifecycleAction::Stop => {
tracing::info!(
box_id = %box_id,
idle_secs = idle.as_secs(),
auto_stop = policy.auto_stop,
"AutoStop deadline reached, stopping box"
);
match tokio::time::timeout(LIFECYCLE_OP_TIMEOUT, state.runtime.get(&box_id)).await {
Ok(Ok(Some(bx))) => {
match tokio::time::timeout(LIFECYCLE_OP_TIMEOUT, bx.stop()).await {
Ok(Ok(())) => {}
Ok(Err(error)) => {
tracing::warn!(box_id = %box_id, %error, "AutoStop failed to stop box");
}
Err(_) => {
tracing::warn!(box_id = %box_id, "AutoStop timed out stopping box");
}
}
}
Ok(Ok(None)) => {}
Ok(Err(error)) => {
tracing::warn!(box_id = %box_id, %error, "AutoStop could not fetch box")
}
Err(_) => {
tracing::warn!(box_id = %box_id, "AutoStop timed out fetching box")
}
}
}
LifecycleAction::Delete => {
tracing::info!(
box_id = %box_id,
at_rest_secs = since_stop.as_secs(),
auto_delete = policy.auto_delete,
"AutoDelete deadline reached, removing box"
);
state.boxes.write().await.remove(&box_id);
match tokio::time::timeout(
LIFECYCLE_OP_TIMEOUT,
state.runtime.remove(&box_id, false),
)
.await
{
Ok(Ok(())) => state.forget_box(&box_id).await,
Ok(Err(error)) => {
tracing::warn!(box_id = %box_id, %error, "AutoDelete failed to remove box");
}
Err(_) => {
tracing::warn!(box_id = %box_id, "AutoDelete timed out removing box");
}
}
}
}
}
}
async fn reaper_loop(state: Arc<AppState>) {
let reconnect_grace = resolve_duration("BOXLITE_RECONNECT_GRACE", DEFAULT_RECONNECT_GRACE);
let shutdown_grace = resolve_duration("BOXLITE_SHUTDOWN_GRACE", DEFAULT_SHUTDOWN_GRACE);
let max_lifetime =
resolve_duration("BOXLITE_MAX_SESSION_LIFETIME", DEFAULT_MAX_SESSION_LIFETIME);
let mut ticker = tokio::time::interval(REAPER_TICK);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
ticker.tick().await;
let now = Instant::now();
run_reap_once(&state, now, reconnect_grace, shutdown_grace, max_lifetime).await;
run_lifecycle_once(&state, now).await;
}
}
async fn run_reap_once(
state: &AppState,
now: Instant,
reconnect_grace: std::time::Duration,
shutdown_grace: std::time::Duration,
max_lifetime: std::time::Duration,
) {
let candidates: Vec<(String, Arc<ActiveExecution>)> = {
let map = state.executions.read().await;
map.iter()
.map(|(k, v)| (k.clone(), Arc::clone(v)))
.collect()
};
for (id, active) in candidates {
if active.is_done() {
if !active.should_retain(now) {
evict_if_same(state, &id, &active).await;
}
continue;
}
if active.kind() == SessionKind::Main {
continue;
}
if now.duration_since(active.created_at()) > max_lifetime {
active.mark_reaping_kill().await;
tracing::warn!(exec_id = %id, "session lifetime cap reached, killing");
try_kill_and_evict(state, &id, &active).await;
continue;
}
if active.is_reaping_kill().await {
try_kill_and_evict(state, &id, &active).await;
continue;
}
if active.try_escalate_hup(now, reconnect_grace).await {
let sig_result =
tokio::time::timeout(REAPER_SIGNAL_TIMEOUT, active.execution().signal(1)).await;
if matches!(sig_result, Ok(Ok(()))) {
active.finish_escalation().await;
} else {
tracing::warn!(exec_id = %id, "SIGHUP delivery failed or timed out, killing");
active.escalation_failed_mark_doomed().await;
try_kill_and_evict(state, &id, &active).await;
}
} else if active.try_escalate_term(now, shutdown_grace).await {
let sig_result =
tokio::time::timeout(REAPER_SIGNAL_TIMEOUT, active.execution().signal(15)).await;
if matches!(sig_result, Ok(Ok(()))) {
active.finish_escalation().await;
} else {
tracing::warn!(exec_id = %id, "SIGTERM delivery failed or timed out, killing");
active.escalation_failed_mark_doomed().await;
try_kill_and_evict(state, &id, &active).await;
}
} else if active.try_escalate_kill(now, shutdown_grace).await {
tracing::warn!(exec_id = %id, "orphan exec did not exit after SIGTERM, killing");
try_kill_and_evict(state, &id, &active).await;
}
}
}
async fn try_kill_and_evict(state: &AppState, id: &str, active: &Arc<ActiveExecution>) {
let result = tokio::time::timeout(REAPER_SIGNAL_TIMEOUT, active.execution().kill()).await;
match result {
Ok(Ok(())) => {
evict_if_same(state, id, active).await;
}
Ok(Err(e)) => {
tracing::warn!(exec_id = %id, err = %e, "kill failed, will retry next tick");
}
Err(_) => {
tracing::warn!(exec_id = %id, "kill timed out, will retry next tick");
}
}
}
async fn evict_if_same(state: &AppState, id: &str, doomed: &Arc<ActiveExecution>) {
let mut map = state.executions.write().await;
if let Some(current) = map.get(id)
&& Arc::ptr_eq(current, doomed)
{
map.remove(id);
}
}
fn box_info_to_response(info: &BoxInfo) -> BoxResponse {
BoxResponse {
box_id: info.id.to_string(),
name: info.name.clone(),
status: info.status.as_str().to_string(),
created_at: info.created_at.to_rfc3339(),
updated_at: info.last_updated.to_rfc3339(),
pid: info.pid,
image: info.image.clone(),
cpus: info.cpus,
memory_mib: info.memory_mib,
labels: info.labels.clone(),
auto_stop: info.auto_stop,
auto_delete: info.auto_delete,
auto_resume: info.auto_resume,
exit_code: info.exit_code,
}
}
fn volume_info_to_response(info: &boxlite::runtime::types::VolumeInfo) -> types::VolumeResponse {
types::VolumeResponse {
id: info.id.clone(),
name: info.name.clone(),
created_at: info.created_at.to_rfc3339(),
size_bytes: info.size_bytes,
}
}
fn build_box_options(req: &CreateBoxRequest) -> Result<BoxOptions, boxlite::BoxliteError> {
let rootfs = if let Some(ref path) = req.rootfs_path {
RootfsSpec::RootfsPath(path.clone())
} else {
RootfsSpec::Image(req.image.clone().unwrap_or_else(|| "alpine:latest".into()))
};
let env: Vec<(String, String)> = req
.env
.as_ref()
.map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
.unwrap_or_default();
let (network, inbound_network) = match &req.network {
Some(network) => {
if network.uses_legacy_fields()
&& (network.outbound.is_some() || network.inbound.is_some())
{
return Err(boxlite::BoxliteError::InvalidArgument(
"network must use either nested outbound/inbound fields or legacy flat fields, not both"
.into(),
));
}
let (mode, allow_net) = match &network.outbound {
Some(outbound) => (
outbound.mode.parse::<NetworkMode>()?,
outbound.allow_net.clone(),
),
None => (
network
.legacy
.mode
.as_deref()
.unwrap_or("enabled")
.parse::<NetworkMode>()?,
network.legacy.allow_net.clone().unwrap_or_default(),
),
};
let (inbound_mode, inbound_allow_net) = match &network.inbound {
Some(inbound) => (
inbound.mode.parse::<NetworkMode>()?,
inbound.allow_net.clone(),
),
None => (NetworkMode::Disabled, Vec::new()),
};
(
NetworkSpec::try_from(OutboundNetworkConfig { mode, allow_net })?,
NetworkSpec::try_from(InboundNetworkConfig {
mode: inbound_mode,
allow_net: inbound_allow_net,
})?,
)
}
None => (NetworkSpec::default(), NetworkSpec::disabled()),
};
if let Some(volumes) = &req.volumes
&& !volumes.is_empty()
{
return Err(boxlite::BoxliteError::InvalidArgument(
"managed volumes are not supported by boxlite serve".into(),
));
}
if let Some(secrets) = &req.secrets
&& secrets
.iter()
.any(|s| s.name.is_empty() || s.value.is_empty())
{
return Err(boxlite::BoxliteError::InvalidArgument(
"secret name and value must be non-empty".into(),
));
}
let secrets: Vec<boxlite::runtime::options::Secret> = req
.secrets
.as_ref()
.map(|ss| {
ss.iter()
.map(|s| boxlite::runtime::options::Secret {
name: s.name.clone(),
value: s.value.clone(),
hosts: s.hosts.clone(),
placeholder: s
.placeholder
.clone()
.filter(|p| !p.is_empty())
.unwrap_or_else(|| format!("<BOXLITE_SECRET:{}>", s.name)),
})
.collect()
})
.unwrap_or_default();
Ok(BoxOptions {
rootfs,
cpus: req.cpus,
memory_mib: req.memory_mib,
disk_size_gb: req.disk_size_gb,
working_dir: req.working_dir.clone(),
env,
secrets,
network,
inbound_network,
entrypoint: req.entrypoint.clone(),
cmd: req.cmd.clone(),
user: req.user.clone(),
tty: req.tty.unwrap_or(false),
advanced: {
let mut advanced = boxlite::AdvancedBoxOptions::default();
let capabilities = req.advanced.capabilities.as_ref().map(|capabilities| {
boxlite::ContainerCapabilities {
add: capabilities.add.clone(),
drop: capabilities.drop.clone(),
}
});
advanced.set_capabilities(capabilities)?;
advanced
},
auto_stop: None,
auto_delete: Some(0),
auto_resume: req.auto_resume,
detach: req.detach.unwrap_or(true),
..Default::default()
})
}
fn build_box_command(req: &ExecRequest) -> Result<BoxCommand, boxlite::BoxliteError> {
let mut cmd = BoxCommand::new(&req.command).args(req.args.iter().map(String::as_str));
if let Some(ref env_map) = req.env {
for (k, v) in env_map {
cmd = cmd.env(k, v);
}
}
if let Some(ref wd) = req.working_dir {
cmd = cmd.working_dir(wd);
}
if req.tty {
cmd = cmd.tty(true);
}
if let Some(secs) = req.timeout_seconds {
cmd = cmd.timeout_seconds(secs)?;
}
Ok(cmd)
}
fn error_response(
status: StatusCode,
message: impl Into<String>,
error_type: &str,
code: &str,
) -> Response {
let body = ErrorBody {
error: ErrorDetail {
message: message.into(),
error_type: error_type.to_string(),
code: code.to_string(),
request_id: None,
},
};
(status, Json(body)).into_response()
}
fn error_from_boxlite(err: &boxlite::BoxliteError) -> Response {
let (code, etype, ecode) = err.http();
let status = StatusCode::from_u16(code)
.expect("BoxliteError::http() must return a valid HTTP status code");
error_response(status, err.to_string(), etype, ecode)
}
fn handle_panic(err: Box<dyn std::any::Any + Send + 'static>) -> Response {
let detail = err
.downcast_ref::<&'static str>()
.map(|s| s.to_string())
.or_else(|| err.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "panic in handler".to_string());
tracing::error!(panic = %detail, "handler panicked");
error_response(
StatusCode::INTERNAL_SERVER_ERROR,
format!("internal error: {}", detail),
"InternalError",
"internal",
)
}
fn auth_allows(expected: Option<&str>, path: &str, bearer: Option<&str>) -> bool {
let Some(expected) = expected else {
return true;
};
if path == "/v1/config" {
return true;
}
match bearer {
Some(tok) => constant_time_eq(tok.as_bytes(), expected.as_bytes()),
None => false,
}
}
fn activity_box_id<'a>(method: &Method, path: &'a str) -> Option<&'a str> {
let rest = path.strip_prefix("/v1/boxes/")?;
let (box_id, tail) = match rest.split_once('/') {
Some((box_id, tail)) => (box_id, tail),
None => (rest, ""),
};
if box_id.is_empty() || matches!(tail, "metrics" | "stop") {
return None;
}
if tail.is_empty() && matches!(*method, Method::GET | Method::HEAD | Method::DELETE) {
return None;
}
Some(box_id)
}
async fn record_activity(State(state): State<Arc<AppState>>, req: Request, next: Next) -> Response {
if let Some(segment) = activity_box_id(req.method(), req.uri().path()) {
let box_id = percent_encoding::percent_decode_str(segment).decode_utf8_lossy();
state.record_box_activity(&box_id).await;
}
next.run(req).await
}
async fn require_api_key(State(state): State<Arc<AppState>>, req: Request, next: Next) -> Response {
let bearer = req
.headers()
.get(AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| {
v.strip_prefix("Bearer ")
.or_else(|| v.strip_prefix("bearer "))
});
if auth_allows(state.api_key.as_deref(), req.uri().path(), bearer) {
next.run(req).await
} else {
error_response(
StatusCode::UNAUTHORIZED,
"invalid or missing API key",
"AuthError",
"unauthenticated",
)
}
}
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
if a.len() != b.len() {
return false;
}
let mut diff = 0u8;
for (x, y) in a.iter().zip(b.iter()) {
diff |= x ^ y;
}
diff == 0
}
fn autoresume_allows(status: BoxStatus, auto_resume: bool) -> bool {
status != BoxStatus::Stopped || auto_resume
}
#[allow(clippy::result_large_err)]
async fn get_or_resume_box(state: &AppState, box_id: &str) -> Result<Arc<LiteBox>, Response> {
let info = match state.runtime.get_info(box_id).await {
Ok(Some(info)) => info,
Ok(None) => {
return Err(error_response(
StatusCode::NOT_FOUND,
format!("box not found: {box_id}"),
"NotFoundError",
"not_found",
));
}
Err(e) => return Err(error_from_boxlite(&e)),
};
if !autoresume_allows(info.status, info.auto_resume) {
return Err(error_response(
StatusCode::CONFLICT,
format!(
"box {box_id} is {:?} and has AutoResume disabled; start it explicitly first",
info.status
),
"ConflictError",
"conflict",
));
}
get_or_fetch_box(state, box_id).await
}
#[allow(clippy::result_large_err)]
async fn get_box_for_metrics(state: &AppState, box_id: &str) -> Result<Arc<LiteBox>, Response> {
let info = match state.runtime.get_info(box_id).await {
Ok(Some(info)) => info,
Ok(None) => {
return Err(error_response(
StatusCode::NOT_FOUND,
format!("box not found: {box_id}"),
"NotFoundError",
"not_found",
));
}
Err(e) => return Err(error_from_boxlite(&e)),
};
let policy = state.lifecycle_of(info.id.as_ref()).await;
if metrics_would_thrash(info.status, policy.auto_stop) {
let status = info.status;
return Err(error_response(
StatusCode::CONFLICT,
format!(
"box {box_id} is {status:?} and AutoStop is enabled; a metrics scrape does not \
start a box, start it explicitly first"
),
"ConflictError",
"conflict",
));
}
get_or_fetch_box(state, box_id).await
}
fn metrics_would_thrash(status: BoxStatus, auto_stop_secs: u32) -> bool {
status != BoxStatus::Running && auto_stop_secs > 0
}
#[allow(clippy::result_large_err)]
async fn get_or_fetch_box(state: &AppState, box_id: &str) -> Result<Arc<LiteBox>, Response> {
let cached = state.boxes.read().await.get(box_id).cloned();
if let Some(cached) = cached {
match cached.info().await {
Ok(info) if info.status.is_active() => return Ok(Arc::clone(&cached)),
Ok(_) => {}
Err(e) => return Err(error_from_boxlite(&e)),
}
let mut boxes = state.boxes.write().await;
if boxes
.get(box_id)
.is_some_and(|current| Arc::ptr_eq(current, &cached))
{
boxes.remove(box_id);
}
}
match state.runtime.get(box_id).await {
Ok(Some(b)) => {
let id = b
.info()
.await
.map_err(|e| error_from_boxlite(&e))?
.id
.to_string();
let arc = Arc::new(b);
state.boxes.write().await.insert(id, Arc::clone(&arc));
Ok(arc)
}
Ok(None) => Err(error_response(
StatusCode::NOT_FOUND,
format!("box not found: {box_id}"),
"NotFoundError",
"not_found",
)),
Err(e) => Err(error_from_boxlite(&e)),
}
}
fn find_main_session(
executions: &HashMap<String, Arc<ActiveExecution>>,
box_id: &str,
) -> Option<Arc<ActiveExecution>> {
executions
.values()
.find(|active| {
active.box_id() == box_id && active.kind() == SessionKind::Main && !active.is_done()
})
.cloned()
}
async fn register_main_session<F, Fut>(
executions: &mut HashMap<String, Arc<ActiveExecution>>,
box_id: &str,
open: F,
) -> Result<Arc<ActiveExecution>, boxlite::BoxliteError>
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = Result<Execution, boxlite::BoxliteError>>,
{
if let Some(active) = find_main_session(executions, box_id) {
return Ok(active);
}
let mut execution = open().await?;
let stdin = execution.stdin();
let exec_id = execution.id().clone();
let active = ActiveExecution::new(box_id.to_string(), SessionKind::Main, execution, stdin);
executions.insert(exec_id, Arc::clone(&active));
Ok(active)
}
#[allow(clippy::result_large_err)]
async fn get_or_attach_main_session(
state: &AppState,
box_id: &str,
) -> Result<Arc<ActiveExecution>, Response> {
if let Some(active) = find_main_session(&*state.executions.read().await, box_id) {
return Ok(active);
}
let litebox = get_or_resume_box(state, box_id).await?;
let mut executions = state.executions.write().await;
register_main_session(&mut executions, box_id, || async {
litebox.attach(boxlite::AttachOptions::main()).await
})
.await
.map_err(|e| error_from_boxlite(&e))
}
fn build_router(state: Arc<AppState>) -> Router {
use handlers::{advanced, boxes, config, executions, files, me, metrics, snapshots, volumes};
Router::new()
.route("/v1/me", get(me::get_me))
.route("/v1/config", get(config::get_config))
.route("/v1/metrics", get(metrics::runtime_metrics))
.route(
"/v1/volumes",
post(volumes::create_volume).get(volumes::list_volumes),
)
.route(
"/v1/volumes/{id}",
get(volumes::get_volume).delete(volumes::remove_volume),
)
.route("/v1/boxes/import", post(advanced::import_box))
.route(
"/v1/boxes",
post(boxes::create_box).get(boxes::list_boxes),
)
.route(
"/v1/boxes/{box_id}",
get(boxes::get_box)
.delete(boxes::remove_box)
.head(boxes::head_box),
)
.route(
"/v1/boxes/{box_id}/start",
post(boxes::start_box),
)
.route(
"/v1/boxes/{box_id}/stop",
post(boxes::stop_box),
)
.route(
"/v1/boxes/{box_id}/metrics",
get(metrics::box_metrics),
)
.route(
"/v1/boxes/{box_id}/attach",
get(executions::attach_box),
)
.route(
"/v1/boxes/{box_id}/exec",
post(executions::start_execution),
)
.route(
"/v1/boxes/{box_id}/executions/{exec_id}",
get(executions::get_execution).delete(executions::kill_execution),
)
.route(
"/v1/boxes/{box_id}/executions/{exec_id}/attach",
get(executions::attach_execution),
)
.route(
"/v1/boxes/{box_id}/executions/{exec_id}/signal",
post(executions::send_signal),
)
.route(
"/v1/boxes/{box_id}/executions/{exec_id}/resize",
post(executions::resize_tty),
)
.route(
"/v1/boxes/{box_id}/files",
put(files::upload_files).get(files::download_files),
)
.route(
"/v1/boxes/{box_id}/snapshots",
post(snapshots::create_snapshot).get(snapshots::list_snapshots),
)
.route(
"/v1/boxes/{box_id}/snapshots/{name}",
get(snapshots::get_snapshot).delete(snapshots::delete_snapshot),
)
.route(
"/v1/boxes/{box_id}/snapshots/{name}/restore",
post(snapshots::restore_snapshot),
)
.route(
"/v1/boxes/{box_id}/clone",
post(advanced::clone_box),
)
.route(
"/v1/boxes/{box_id}/export",
post(advanced::export_box),
)
.layer(middleware::from_fn_with_state(
state.clone(),
record_activity,
))
.layer(middleware::from_fn_with_state(
state.clone(),
require_api_key,
))
.layer(CatchPanicLayer::custom(handle_panic))
.layer(PropagateRequestIdLayer::x_request_id())
.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
.with_state(state)
}
pub async fn execute(args: ServeArgs, global: &GlobalFlags) -> anyhow::Result<()> {
let options = global.resolve_runtime_options()?;
let runtime = global.create_runtime_with_options(options)?;
let state = Arc::new(AppState {
runtime,
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: args.api_key.clone(),
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
tokio::spawn(reaper_loop(Arc::clone(&state)));
let app = build_router(state.clone());
let listener = tokio::net::TcpListener::bind((args.host.as_str(), args.port)).await?;
let addr = listener.local_addr()?;
tracing::info!("boxlite serve listening on {}", addr);
eprintln!("BoxLite REST API server listening on http://{addr}");
let shutdown_state = state.clone();
axum::serve(listener, app)
.with_graceful_shutdown(async move {
let _ = tokio::signal::ctrl_c().await;
tracing::info!("shutting down...");
eprintln!("\nShutting down...");
let _ = shutdown_state.runtime.shutdown(Some(10)).await;
})
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use boxlite::runtime::options::NetworkSpec;
use std::time::Duration;
#[test]
fn auth_allows_permissive_when_no_key() {
assert!(auth_allows(None, "/v1/boxes", None));
assert!(auth_allows(None, "/v1/me", Some("anything")));
}
#[test]
fn auth_allows_config_public_even_with_key() {
assert!(auth_allows(Some("k"), "/v1/config", None));
}
#[test]
fn auth_allows_requires_exact_bearer_when_key_set() {
assert!(auth_allows(Some("k"), "/v1/me", Some("k")));
assert!(!auth_allows(Some("k"), "/v1/me", Some("wrong")));
assert!(!auth_allows(Some("k"), "/v1/me", None));
assert!(!auth_allows(Some("k"), "/v1/boxes", Some("")));
}
#[test]
fn constant_time_eq_basic() {
assert!(constant_time_eq(b"abc", b"abc"));
assert!(!constant_time_eq(b"abc", b"abd"));
assert!(!constant_time_eq(b"abc", b"abcd"));
assert!(constant_time_eq(b"", b""));
}
#[test]
fn build_box_command_rejects_invalid_timeout_seconds() {
for seconds in [-1.0, f64::NAN, f64::INFINITY] {
let req = ExecRequest {
command: "true".to_string(),
args: Vec::new(),
stdin: None,
env: None,
timeout_seconds: Some(seconds),
working_dir: None,
tty: false,
};
let err = build_box_command(&req).expect_err("invalid timeout should fail");
assert!(
matches!(err, boxlite::BoxliteError::InvalidArgument(ref msg) if msg.contains("timeout_seconds")),
"unexpected error for {seconds:?}: {err}"
);
}
}
#[test]
fn build_box_options_carries_tty_from_the_wire() {
let with_tty: super::types::CreateBoxRequest =
serde_json::from_str(r#"{"image": "alpine:latest", "tty": true}"#)
.expect("body with tty must deserialize");
assert!(
build_box_options(&with_tty).expect("build").tty,
"a REST client asking for a terminal must get one"
);
let without: super::types::CreateBoxRequest =
serde_json::from_str(r#"{"image": "alpine:latest"}"#).expect("body must deserialize");
assert!(
!build_box_options(&without).expect("build").tty,
"no tty asked for, none granted"
);
}
#[test]
fn build_box_options_carries_secrets_from_the_wire() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{"image":"alpine:latest","secrets":[{"name":"openai","value":"sk-test","hosts":["api.openai.com"]}]}"#,
)
.expect("body with secrets must deserialize");
let opts = build_box_options(&req).expect("build with secrets");
assert_eq!(opts.secrets.len(), 1, "one secret in, one secret out");
let secret = &opts.secrets[0];
assert_eq!(secret.name, "openai");
assert_eq!(secret.value, "sk-test");
assert_eq!(secret.hosts, vec!["api.openai.com"]);
assert_eq!(secret.placeholder, "<BOXLITE_SECRET:openai>");
}
#[test]
fn build_box_options_preserves_explicit_secret_placeholder() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{"image":"alpine:latest","secrets":[{"name":"httpbin","value":"v","placeholder":"<MY_TOKEN>"}]}"#,
)
.expect("body with explicit placeholder must deserialize");
let opts = build_box_options(&req).expect("build");
assert_eq!(
opts.secrets[0].placeholder, "<MY_TOKEN>",
"caller placeholder wins"
);
}
#[test]
fn build_box_options_defaults_an_empty_secret_placeholder() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{"image":"alpine:latest","secrets":[{"name":"openai","value":"v","placeholder":""}]}"#,
)
.expect("body with empty placeholder must deserialize");
let opts = build_box_options(&req).expect("build");
assert_eq!(
opts.secrets[0].placeholder, "<BOXLITE_SECRET:openai>",
"an empty placeholder is as absent as an omitted one"
);
}
#[test]
fn build_box_options_rejects_an_empty_secret_name_or_value() {
for secrets in [
r#"[{"name":"","value":"v"}]"#,
r#"[{"name":"n","value":""}]"#,
] {
let req: super::types::CreateBoxRequest = serde_json::from_str(&format!(
r#"{{"image":"alpine:latest","secrets":{secrets}}}"#
))
.expect("body with secrets must deserialize");
let err = build_box_options(&req).expect_err("empty secret fields must be rejected");
assert!(
matches!(err, boxlite::BoxliteError::InvalidArgument(ref msg) if msg.contains("non-empty")),
"unexpected error: {err}"
);
}
}
#[test]
fn build_box_options_rejects_nonempty_volumes() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{"image":"alpine:latest","volumes":[{"managed_volume":"v1","guest_path":"/data"}]}"#,
)
.expect("body with volumes must deserialize (accepted, then rejected)");
let err = build_box_options(&req).expect_err("non-empty volumes must be rejected");
assert!(
matches!(err, boxlite::BoxliteError::InvalidArgument(ref msg) if msg.contains("managed volumes")),
"unexpected error: {err}"
);
}
#[test]
fn build_box_options_carries_container_capabilities_from_the_wire() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{"image":"alpine:latest","advanced":{"capabilities":{"add":["SYS_ADMIN"],"drop":["CAP_NET_RAW"]}}}"#,
)
.expect("capability request must deserialize");
let opts = build_box_options(&req).expect("build capability options");
let capabilities = opts.advanced.capabilities().expect("capabilities set");
assert_eq!(capabilities.add, vec!["SYS_ADMIN"]);
assert_eq!(capabilities.drop, vec!["CAP_NET_RAW"]);
}
#[test]
fn build_box_options_leaves_capabilities_unspecified_when_the_wire_omits_them() {
let req: super::types::CreateBoxRequest =
serde_json::from_str(r#"{"image":"alpine:latest"}"#)
.expect("ordinary request must deserialize");
let opts = build_box_options(&req).expect("build ordinary options");
assert!(
opts.advanced.capabilities().is_none(),
"omitting capabilities on the wire must not become an explicit empty policy"
);
}
#[test]
fn build_box_options_carries_lifecycle_policy_and_uses_compatible_detach_default() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{"auto_stop": 900, "auto_delete": 3600, "auto_resume": false}"#,
)
.expect("lifecycle body must deserialize");
let opts = build_box_options(&req).expect("build lifecycle options");
assert_eq!(
opts.auto_stop, None,
"the local runtime refuses a non-zero auto_stop outright, so \
forwarding it would 400 the create"
);
assert_eq!(opts.auto_resume, Some(false));
assert_eq!(
opts.auto_delete,
Some(0),
"the deadline must not reach the runtime, which would read any \
non-zero value as remove-on-stop and delete the box at its stop"
);
assert!(
opts.detach,
"withholding the deadline is what keeps a deadlined box detachable"
);
let persistent: super::types::CreateBoxRequest =
serde_json::from_str(r#"{"auto_delete": 0}"#).expect("body must deserialize");
assert!(
build_box_options(&persistent).expect("build").detach,
"persistent boxes keep the serve API's historical detached default"
);
}
fn lifecycle_state() -> AppState {
let runtime = BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(
"http://127.0.0.1:1".to_string(),
))
.expect("rest runtime");
AppState {
runtime,
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
}
}
fn info_as_the_runtime_reports_it(status: &str) -> BoxInfo {
serde_json::from_value(serde_json::json!({
"id": "01HJK4TNRPQSXYZ8WM6NCVT9R5",
"name": "swept",
"status": status,
"created_at": "2026-01-01T00:00:00Z",
"last_updated": "2026-01-01T00:00:00Z",
"pid": null,
"image": "alpine:latest",
"cpus": 1,
"memory_mib": 512,
"labels": {},
"auto_stop": 0,
"auto_delete": 0,
"auto_resume": true,
"health_status": { "state": "None", "failures": 0, "last_check": null },
"exit_code": null,
}))
.expect("box info")
}
#[tokio::test]
async fn a_box_response_reports_the_deadlines_serve_holds() {
let state = lifecycle_state();
let info = info_as_the_runtime_reports_it("stopped");
assert_eq!(
(info.auto_stop, info.auto_delete),
(0, 0),
"the runtime has no deadline to report; the whole point is that serve holds it"
);
state
.set_lifecycle(
info.id.as_ref(),
LifecyclePolicy {
auto_stop: 900,
auto_delete: 604_800,
},
)
.await;
let resp = state.box_response(&info).await;
assert_eq!(resp.auto_stop, 900, "AutoStop must survive the round trip");
assert_eq!(
resp.auto_delete, 604_800,
"AutoDelete must survive the round trip"
);
assert!(
resp.auto_resume,
"auto_resume is forwarded to the runtime, so it is read off the info"
);
}
#[tokio::test]
async fn a_box_with_no_stored_policy_reports_no_deadline() {
let state = lifecycle_state();
let resp = state
.box_response(&info_as_the_runtime_reports_it("running"))
.await;
assert_eq!((resp.auto_stop, resp.auto_delete), (0, 0));
}
#[test]
fn metrics_refuses_to_wake_only_a_box_the_sweep_would_stop() {
for status in [
BoxStatus::Unknown,
BoxStatus::Configured,
BoxStatus::Stopping,
BoxStatus::Stopped,
BoxStatus::Paused,
BoxStatus::Failed,
] {
assert!(
metrics_would_thrash(status, 900),
"{status:?} is not running, so serving metrics would boot it into the sweep"
);
assert!(
!metrics_would_thrash(status, 0),
"{status:?} has no AutoStop deadline, so nothing would stop it again"
);
}
assert!(
!metrics_would_thrash(BoxStatus::Running, 900),
"a running box is not booted by a scrape, so there is no wake to refuse"
);
assert!(!metrics_would_thrash(BoxStatus::Running, 0));
}
#[tokio::test]
async fn an_invalid_lifecycle_pair_is_refused_before_the_box_is_created() {
let req: types::CreateBoxRequest =
serde_json::from_str(r#"{"auto_stop": 3600, "auto_delete": 60}"#)
.expect("body must deserialize");
let response = handlers::boxes::create_box(
axum::extract::State(Arc::new(lifecycle_state())),
axum::Json(req),
)
.await;
assert_eq!(
response.status(),
StatusCode::BAD_REQUEST,
"auto_delete <= auto_stop is forbidden by the contract"
);
}
#[tokio::test]
async fn only_a_tenant_exec_marks_a_box_busy() {
let (main_exec, _main_ch) = stub_execution("cid-main");
let (tenant, _tenant_ch) = stub_execution("cid-exec");
let mut executions: HashMap<String, Arc<ActiveExecution>> = HashMap::new();
executions.insert(
"main".to_string(),
ActiveExecution::new("box-main".to_string(), SessionKind::Main, main_exec, None),
);
executions.insert(
"exec".to_string(),
ActiveExecution::new("box-exec".to_string(), SessionKind::Exec, tenant, None),
);
let busy = busy_box_ids(&executions);
assert!(busy.contains("box-exec"), "a live tenant exec is work");
assert!(
!busy.contains("box-main"),
"counting Main would make AutoStop unreachable for every started box"
);
let mut dones: Vec<_> = executions.values().map(|active| active.done_rx()).collect();
drop((_main_ch, _tenant_ch));
for done in &mut dones {
assert!(
tokio::time::timeout(std::time::Duration::from_secs(5), done.changed())
.await
.is_ok(),
"a session whose channels are closed must finish"
);
}
}
#[test]
fn a_spelling_resolves_the_way_the_runtime_resolves_it() {
let id = "01HJK4TNRPQSXYZ8WM6NCVT9R5";
let other = "01HJK9ZZZZZZZZZZZZZZZZZZZZ";
let one = [(id.to_string(), Some("web".to_string()))];
let two = [
(id.to_string(), Some("web".to_string())),
(other.to_string(), None),
];
assert_eq!(box_named_by(id, &one), Some(id), "exact id");
assert_eq!(box_named_by("web", &one), Some(id), "exact name");
assert_eq!(box_named_by("01HJK4", &two), Some(id), "unique prefix");
assert_eq!(box_named_by("", &one), None, "an empty spelling names none");
assert_eq!(box_named_by("nope", &one), None, "names no box");
assert_eq!(
box_named_by("01HJK4TNRPQSXYZ8WM6NCVT9R5X", &one),
None,
"longer than the id is not a prefix of it"
);
assert_eq!(
box_named_by("01HJK", &two),
None,
"an ambiguous prefix names no box"
);
assert_eq!(
box_named_by("0", &two),
None,
"nor does a one-character one"
);
}
#[tokio::test]
async fn a_box_driven_by_an_id_prefix_is_not_idle_under_its_full_id() {
let map: RwLock<HashMap<String, Instant>> = RwLock::new(HashMap::new());
let now = Instant::now();
map.write().await.insert("01HJK4".to_string(), now);
let idle = tokio::time::timeout(
std::time::Duration::from_secs(3),
idle_or_seed(
&map,
"01HJK4TNRPQSXYZ8WM6NCVT9R5",
&[("01HJK4TNRPQSXYZ8WM6NCVT9R5".to_string(), None)],
now + std::time::Duration::from_secs(30),
),
)
.await
.expect("reading must not deadlock");
assert_eq!(
idle,
std::time::Duration::from_secs(30),
"a prefix-addressed request is use of the box it resolves to"
);
}
#[tokio::test]
async fn pruning_keeps_a_stamp_made_under_an_id_prefix() {
let state = lifecycle_state();
state.record_box_activity("01HJK4").await;
state
.retain_known_boxes(&[("01HJK4TNRPQSXYZ8WM6NCVT9R5".to_string(), None)])
.await;
assert!(
state.last_activity.read().await.contains_key("01HJK4"),
"pruning dropped a stamp that names a live box"
);
}
#[tokio::test]
async fn a_box_driven_by_name_is_not_idle_under_its_id() {
let map: RwLock<HashMap<String, Instant>> = RwLock::new(HashMap::new());
let now = Instant::now();
map.write().await.insert("web".to_string(), now);
let idle = tokio::time::timeout(
std::time::Duration::from_secs(3),
idle_or_seed(
&map,
"bx-abc123",
&[("bx-abc123".to_string(), Some("web".to_string()))],
now + std::time::Duration::from_secs(30),
),
)
.await
.expect("reading must not deadlock");
assert_eq!(idle, std::time::Duration::from_secs(30));
}
#[tokio::test]
async fn the_idle_clock_drops_spellings_that_name_no_box() {
let state = lifecycle_state();
state.record_box_activity("real").await;
state.record_box_activity("never-existed").await;
state
.retain_known_boxes(&[("real".to_string(), None)])
.await;
let clocks = state.last_activity.read().await;
assert!(clocks.contains_key("real"));
assert!(
!clocks.contains_key("never-existed"),
"an id no box answers to must not hold a clock forever"
);
}
#[tokio::test]
async fn pruning_never_drops_a_deadline() {
let state = lifecycle_state();
state
.set_lifecycle(
"fresh",
LifecyclePolicy {
auto_stop: 60,
auto_delete: 0,
},
)
.await;
state.retain_known_boxes(&[]).await;
assert_eq!(state.lifecycle_of("fresh").await.auto_stop, 60);
}
#[test]
fn autoresume_gates_only_a_stopped_box() {
assert!(autoresume_allows(BoxStatus::Stopped, true));
assert!(
!autoresume_allows(BoxStatus::Stopped, false),
"a stopped box must refuse an implicit wake"
);
for status in [
BoxStatus::Running,
BoxStatus::Paused,
BoxStatus::Configured,
BoxStatus::Failed,
BoxStatus::Stopping,
BoxStatus::Unknown,
] {
for auto_resume in [true, false] {
assert!(
autoresume_allows(status, auto_resume),
"{status:?} with auto_resume={auto_resume} is not a resume"
);
}
}
}
#[test]
fn a_box_is_stopped_only_after_a_full_idle_window() {
use std::time::Duration;
let below = decide_lifecycle(
BoxStatus::Running,
60,
0,
Duration::from_secs(59),
Duration::ZERO,
);
assert_eq!(below, LifecycleAction::Leave);
let at = decide_lifecycle(
BoxStatus::Running,
60,
0,
Duration::from_secs(60),
Duration::ZERO,
);
assert_eq!(at, LifecycleAction::Stop);
}
#[test]
fn a_zero_window_disables_its_deadline() {
use std::time::Duration;
let huge = Duration::from_secs(86_400);
assert_eq!(
decide_lifecycle(BoxStatus::Running, 0, 0, huge, huge),
LifecycleAction::Leave
);
assert_eq!(
decide_lifecycle(BoxStatus::Stopped, 0, 0, huge, huge),
LifecycleAction::Leave
);
}
#[test]
fn a_box_at_rest_is_deleted_on_its_own_deadline() {
use std::time::Duration;
for status in [BoxStatus::Stopped, BoxStatus::Failed] {
assert_eq!(
decide_lifecycle(status, 0, 60, Duration::ZERO, Duration::from_secs(60)),
LifecycleAction::Delete,
"status {status:?}"
);
}
}
#[test]
fn a_running_box_is_never_deleted_by_the_delete_deadline() {
use std::time::Duration;
assert_eq!(
decide_lifecycle(
BoxStatus::Running,
0,
60,
Duration::ZERO,
Duration::from_secs(86_400)
),
LifecycleAction::Leave
);
}
#[test]
fn box_scoped_paths_count_as_activity_except_the_pollers() {
let post = Method::POST;
assert_eq!(activity_box_id(&post, "/v1/boxes/abc/exec"), Some("abc"));
assert_eq!(activity_box_id(&post, "/v1/boxes/abc/files"), Some("abc"));
assert_eq!(activity_box_id(&post, "/v1/boxes/abc/attach"), Some("abc"));
assert_eq!(activity_box_id(&post, "/v1/boxes/abc/start"), Some("abc"));
assert_eq!(
activity_box_id(&Method::GET, "/v1/boxes/abc/files"),
Some("abc")
);
assert_eq!(activity_box_id(&Method::GET, "/v1/boxes/abc"), None);
assert_eq!(activity_box_id(&Method::HEAD, "/v1/boxes/abc"), None);
assert_eq!(activity_box_id(&Method::DELETE, "/v1/boxes/abc"), None);
assert_eq!(activity_box_id(&post, "/v1/boxes/abc/metrics"), None);
assert_eq!(activity_box_id(&post, "/v1/boxes/abc/stop"), None);
assert_eq!(activity_box_id(&post, "/v1/metrics"), None);
assert_eq!(activity_box_id(&post, "/v1/boxes"), None);
}
#[tokio::test]
async fn an_unseen_box_starts_its_window_now_rather_than_looking_idle() {
let map: RwLock<HashMap<String, Instant>> = RwLock::new(HashMap::new());
let now = Instant::now();
let budget = std::time::Duration::from_secs(3);
let first = tokio::time::timeout(
budget,
idle_or_seed(&map, "b", &[("b".to_string(), None)], now),
)
.await
.expect("seeding must not deadlock");
assert_eq!(first, std::time::Duration::ZERO);
let later = now + std::time::Duration::from_secs(600);
let second = tokio::time::timeout(
budget,
idle_or_seed(&map, "b", &[("b".to_string(), None)], later),
)
.await
.expect("reading a seeded box must not deadlock");
assert_eq!(second, std::time::Duration::from_secs(600));
}
#[test]
fn build_box_options_omitted_inbound_defaults_to_disabled() {
for network in [
None,
Some(r#"{"mode": "enabled"}"#),
Some(r#"{"outbound": {"mode": "enabled"}}"#),
] {
let body = match network {
Some(network) => format!(r#"{{"image": "alpine:latest", "network": {network}}}"#),
None => r#"{"image": "alpine:latest"}"#.to_string(),
};
let req: super::types::CreateBoxRequest =
serde_json::from_str(&body).expect("body must deserialize");
let opts = build_box_options(&req).expect("build");
assert!(
matches!(opts.inbound_network, NetworkSpec::Disabled),
"network {network:?} must default inbound to disabled"
);
}
}
#[test]
fn build_box_options_accepts_nested_network_spec() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{
"image": "alpine:latest",
"network": {
"outbound": {
"mode": "enabled",
"allow_net": ["api.openai.com"]
},
"inbound": {
"mode": "disabled"
}
}
}"#,
)
.expect("nested network body must deserialize");
let opts = build_box_options(&req).expect("build");
match opts.network {
NetworkSpec::Enabled { allow_net } => {
assert_eq!(allow_net, vec!["api.openai.com"]);
}
NetworkSpec::Disabled => panic!("network should be enabled"),
}
assert!(matches!(opts.inbound_network, NetworkSpec::Disabled));
}
#[test]
fn build_box_options_rejects_mixed_legacy_and_nested_network_spec() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{
"image": "alpine:latest",
"network": {
"mode": "enabled",
"outbound": {
"mode": "enabled",
"allow_net": ["api.openai.com"]
}
}
}"#,
)
.expect("mixed network body still deserializes for compatibility validation");
let err = build_box_options(&req).expect_err("mixed network body must fail");
assert!(
matches!(err, boxlite::BoxliteError::InvalidArgument(ref msg) if msg.contains("either nested")),
"unexpected error: {err}"
);
}
#[test]
fn build_box_options_rejects_empty_legacy_allow_net_mixed_with_nested_network_spec() {
let req: super::types::CreateBoxRequest = serde_json::from_str(
r#"{
"image": "alpine:latest",
"network": {
"allow_net": [],
"outbound": {
"mode": "enabled"
}
}
}"#,
)
.expect("mixed network body still deserializes for compatibility validation");
let err = build_box_options(&req).expect_err("mixed network body must fail");
assert!(
matches!(err, boxlite::BoxliteError::InvalidArgument(ref msg) if msg.contains("either nested")),
"unexpected error: {err}"
);
}
#[test]
fn build_box_options_empty_body_lands_on_server_default_security() {
let json = r#"{"image": "alpine:latest"}"#;
let req: super::types::CreateBoxRequest =
serde_json::from_str(json).expect("body must deserialize");
let opts = build_box_options(&req).expect("build_box_options");
#[cfg(any(target_os = "linux", target_os = "macos"))]
assert!(
opts.advanced.security.jailer_enabled,
"server default must be sandbox-on after the flip"
);
}
#[test]
fn create_box_request_rejects_client_supplied_security_preset() {
let json = r#"{"image": "alpine:latest", "security": "development"}"#;
let msg = match serde_json::from_str::<super::types::CreateBoxRequest>(json) {
Ok(_) => panic!("`security` must be rejected at deserialize"),
Err(e) => e.to_string(),
};
assert!(
msg.contains("unknown field") && msg.contains("security"),
"expected deny-unknown-fields rejection mentioning `security`; got {msg}"
);
}
#[test]
fn create_box_request_rejects_client_supplied_security_settings() {
let json = r#"{
"image": "alpine:latest",
"security_settings": {
"jailer_enabled": false,
"seccomp_enabled": false,
"uid": null,
"gid": null,
"new_pid_ns": false,
"new_net_ns": false,
"chroot_base": "/srv/boxlite",
"chroot_enabled": false,
"close_fds": false,
"sanitize_env": false,
"env_allowlist": [],
"resource_limits": {},
"sandbox_profile": null,
"network_enabled": true
}
}"#;
let msg = match serde_json::from_str::<super::types::CreateBoxRequest>(json) {
Ok(_) => panic!("`security_settings` must be rejected at deserialize"),
Err(e) => e.to_string(),
};
assert!(
msg.contains("unknown field") && msg.contains("security_settings"),
"expected deny-unknown-fields rejection mentioning `security_settings`; got {msg}"
);
}
fn uploaded_v3_archive(box_options: serde_json::Value) -> Vec<u8> {
let backing_path = b"/server/path/must-not-be-read";
let mut disk = vec![0_u8; 1024];
disk[0..4].copy_from_slice(&0x5146_49fbu32.to_be_bytes());
disk[4..8].copy_from_slice(&3_u32.to_be_bytes());
disk[8..16].copy_from_slice(&512_u64.to_be_bytes());
disk[16..20].copy_from_slice(&(backing_path.len() as u32).to_be_bytes());
disk[512..512 + backing_path.len()].copy_from_slice(backing_path);
let manifest = serde_json::json!({
"version": 3,
"box_name": null,
"image": "alpine:latest",
"box_options": box_options,
"guest_disk_checksum": "",
"container_disk_checksum": "",
"exported_at": "2026-07-26T00:00:00Z"
});
let manifest = serde_json::to_vec(&manifest).expect("serialize manifest");
let mut archive = tar::Builder::new(Vec::new());
for (name, bytes) in [
("manifest.json", manifest.as_slice()),
("disk.qcow2", disk.as_slice()),
] {
let mut header = tar::Header::new_gnu();
header.set_mode(0o600);
header.set_size(bytes.len() as u64);
header.set_cksum();
archive
.append_data(&mut header, name, bytes)
.expect("append archive entry");
}
archive.finish().expect("finish archive");
archive.into_inner().expect("archive bytes")
}
async fn upload_v3_archive(
state: Arc<AppState>,
box_options: serde_json::Value,
) -> (StatusCode, serde_json::Value) {
let response = handlers::advanced::import_box(
State(state),
axum::extract::Query(types::ImportQuery { name: None }),
axum::body::Bytes::from(uploaded_v3_archive(box_options)),
)
.await;
let status = response.status();
let response_body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.expect("response body");
let body = serde_json::from_slice(&response_body).expect("JSON response");
(status, body)
}
async fn assert_uploaded_archive_rejected_before_provisioning(
box_options: serde_json::Value,
expected_message: &str,
) {
let home = tempfile::tempdir().expect("runtime home");
let runtime = BoxliteRuntime::new_for_test(boxlite::BoxliteOptions {
home_dir: home.path().join("boxlite"),
..Default::default()
})
.expect("local runtime");
let state = Arc::new(AppState {
runtime: runtime.clone(),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
let (control_status, control_error) =
upload_v3_archive(state.clone(), serde_json::json!({})).await;
assert_eq!(control_status, StatusCode::CONFLICT);
assert_eq!(control_error["error"]["type"], "InvalidStateError");
assert_eq!(control_error["error"]["code"], "invalid_state");
assert!(
control_error["error"]["message"]
.as_str()
.is_some_and(|message| message.contains("backing file reference")),
"control upload must reach disk validation: {control_error}"
);
let (status, error) = upload_v3_archive(state, box_options).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(error["error"]["type"], "UnsupportedError");
assert_eq!(error["error"]["code"], "unsupported");
assert!(
error["error"]["message"]
.as_str()
.is_some_and(|message| message.contains(expected_message))
);
assert!(
runtime.list_info().await.expect("list boxes").is_empty(),
"rejected upload must not provision a box"
);
runtime.shutdown(Some(1)).await.expect("shutdown runtime");
}
#[tokio::test]
async fn serve_import_rejects_nested_virtualization_archive_before_provisioning() {
assert_uploaded_archive_rejected_before_provisioning(
serde_json::json!({"advanced": {"nested_virtualization": true}}),
"nested virtualization",
)
.await;
}
#[tokio::test]
async fn serve_import_rejects_host_volume_archive_before_provisioning() {
assert_uploaded_archive_rejected_before_provisioning(
serde_json::json!({
"volumes": [{
"host_path": "/",
"guest_path": "/host",
"read_only": false
}]
}),
"volume mounts",
)
.await;
}
#[tokio::test]
async fn serve_import_rejects_managed_volume_archive_before_provisioning() {
assert_uploaded_archive_rejected_before_provisioning(
serde_json::json!({
"volumes": [{
"managed_volume": "someone-elses-data",
"guest_path": "/data",
"read_only": false
}]
}),
"volume mounts",
)
.await;
}
fn make_test_active() -> (
Arc<ActiveExecution>,
tokio::sync::mpsc::UnboundedSender<String>, // stdout driver
tokio::sync::mpsc::UnboundedSender<String>, // stderr driver
tokio::sync::mpsc::UnboundedSender<boxlite::ExecResult>, // result driver
) {
let (exec, stdout_tx, stderr_tx, _stdin_rx, result_tx) =
boxlite::Execution::stub("test-exec");
let active = ActiveExecution::new("test-box".to_string(), SessionKind::Exec, exec, None);
(active, stdout_tx, stderr_tx, result_tx)
}
#[allow(dead_code)]
struct StubChannels(
tokio::sync::mpsc::UnboundedSender<String>,
tokio::sync::mpsc::UnboundedSender<String>,
tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
tokio::sync::mpsc::UnboundedSender<boxlite::ExecResult>,
);
fn stub_execution(id: &str) -> (Execution, StubChannels) {
let (exec, stdout_tx, stderr_tx, stdin_rx, result_tx) = boxlite::Execution::stub(id);
(
exec,
StubChannels(stdout_tx, stderr_tx, stdin_rx, result_tx),
)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn second_attach_reuses_main_session_and_never_reopens_it() {
let mut executions: HashMap<String, Arc<ActiveExecution>> = HashMap::new();
let opens = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let (tenant, _tenant_channels) = stub_execution("exec-1");
executions.insert(
"exec-1".to_string(),
ActiveExecution::new("box1".to_string(), SessionKind::Exec, tenant, None),
);
let (init, _init_channels) = stub_execution("container-1");
let mut init = Some(init);
let first = register_main_session(&mut executions, "box1", || {
opens.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let init = init.take().expect("opener runs once");
async move { Ok(init) }
})
.await
.expect("first attach opens the main session");
let (decoy, _decoy_channels) = stub_execution("container-DECOY");
let mut decoy = Some(decoy);
let second = register_main_session(&mut executions, "box1", || {
opens.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let decoy = decoy.take().expect("opener runs once");
async move { Ok(decoy) }
})
.await
.expect("second attach resolves the main session");
assert_eq!(
opens.load(std::sync::atomic::Ordering::SeqCst),
1,
"the second attach must reuse the registered main session, not open another one",
);
assert!(
Arc::ptr_eq(&first, &second),
"both attaches must resolve to the same ActiveExecution",
);
assert_eq!(
second.execution().id(),
"container-1",
"the main session keeps the container id it was opened with",
);
assert!(
executions.contains_key("container-1"),
"the main session is registered under its container id, alongside tenant execs: {:?}",
executions.keys().collect::<Vec<_>>(),
);
assert!(first.mark_connected().await, "first client claims the slot");
assert!(
!second.mark_connected().await,
"a second client on an attached main session must be refused (409)",
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn find_main_session_matches_only_the_boxs_init_session() {
let mut executions: HashMap<String, Arc<ActiveExecution>> = HashMap::new();
let (exec, _exec_channels) = stub_execution("exec-1");
executions.insert(
"exec-1".to_string(),
ActiveExecution::new("box1".to_string(), SessionKind::Exec, exec, None),
);
let (other_init, _other_channels) = stub_execution("container-2");
executions.insert(
"container-2".to_string(),
ActiveExecution::new("box2".to_string(), SessionKind::Main, other_init, None),
);
assert!(
find_main_session(&executions, "box1").is_none(),
"box1 has only an exec session — attaching must open its main session, not adopt the exec",
);
let (init, _init_channels) = stub_execution("container-1");
executions.insert(
"container-1".to_string(),
ActiveExecution::new("box1".to_string(), SessionKind::Main, init, None),
);
let found = find_main_session(&executions, "box1").expect("box1 main session");
assert_eq!(
found.execution().id(),
"container-1",
"must not return box2's main session",
);
}
const STUB_BOX_ID: &str = "01HJK4TNRPQSXYZ8WM6NCVT9R5";
const STUB_BOX_NAME: &str = "swept-box";
struct StubUpstream {
url: String,
task: tokio::task::JoinHandle<()>,
stops: Arc<std::sync::atomic::AtomicUsize>,
deletes: Arc<std::sync::atomic::AtomicUsize>,
execs: Arc<std::sync::atomic::AtomicUsize>,
}
impl StubUpstream {
fn stops(&self) -> usize {
self.stops.load(std::sync::atomic::Ordering::SeqCst)
}
fn deletes(&self) -> usize {
self.deletes.load(std::sync::atomic::Ordering::SeqCst)
}
fn execs(&self) -> usize {
self.execs.load(std::sync::atomic::Ordering::SeqCst)
}
}
async fn stub_upstream(status: &'static str, auto_resume: bool) -> StubUpstream {
fn canned_box(status: &str, auto_resume: bool) -> serde_json::Value {
serde_json::json!({
"box_id": STUB_BOX_ID,
"name": STUB_BOX_NAME,
"status": status,
"created_at": "2026-01-01T00:00:00Z",
"updated_at": "2026-01-01T00:00:00Z",
"pid": null,
"image": "alpine:latest",
"cpus": 1,
"memory_mib": 512,
"auto_stop": 0,
"auto_delete": 0,
"auto_resume": auto_resume,
})
}
let deletes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let stops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let execs = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let app = axum::Router::new()
.route(
"/v1/boxes/{id}/exec",
axum::routing::post({
let execs = Arc::clone(&execs);
move |axum::extract::Path(id): axum::extract::Path<String>| {
let execs = Arc::clone(&execs);
async move {
if id == STUB_BOX_ID || id == STUB_BOX_NAME {
execs.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
}),
)
.route(
"/v1/boxes",
axum::routing::get(move || async move {
Json(serde_json::json!({ "boxes": [canned_box(status, auto_resume)] }))
})
.post(move |_body: Json<serde_json::Value>| async move {
(StatusCode::CREATED, Json(canned_box(status, auto_resume)))
}),
)
.route(
"/v1/boxes/{id}/stop",
axum::routing::post({
let stops = Arc::clone(&stops);
move |axum::extract::Path(id): axum::extract::Path<String>| {
let stops = Arc::clone(&stops);
async move {
if id == STUB_BOX_ID || id == STUB_BOX_NAME {
stops.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Json(canned_box("stopped", auto_resume)).into_response()
} else {
StatusCode::NOT_FOUND.into_response()
}
}
}
}),
)
.route(
"/v1/boxes/{id}",
axum::routing::get(
move |axum::extract::Path(id): axum::extract::Path<String>| async move {
if id == STUB_BOX_ID || id == STUB_BOX_NAME {
Json(canned_box(status, auto_resume)).into_response()
} else {
StatusCode::NOT_FOUND.into_response()
}
},
)
.delete({
let deletes = Arc::clone(&deletes);
move |axum::extract::Path(id): axum::extract::Path<String>| {
let deletes = Arc::clone(&deletes);
async move {
if id == STUB_BOX_ID || id == STUB_BOX_NAME {
deletes.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
StatusCode::NO_CONTENT.into_response()
} else {
StatusCode::NOT_FOUND.into_response()
}
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let task = tokio::spawn(async move {
let _ = axum::serve(listener, app).await;
});
StubUpstream {
url: format!("http://127.0.0.1:{port}"),
task,
stops,
deletes,
execs,
}
}
async fn serve_router(state: Arc<AppState>) -> (String, tokio::task::JoinHandle<()>) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let handle = tokio::spawn(async move {
let _ = axum::serve(listener, build_router(state)).await;
});
(format!("http://127.0.0.1:{port}"), handle)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reading_a_box_over_http_reports_the_deadlines_serve_holds() {
let upstream = stub_upstream("stopped", true).await;
let state = Arc::new(AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
state
.set_lifecycle(
STUB_BOX_ID,
LifecyclePolicy {
auto_stop: 900,
auto_delete: 604_800,
},
)
.await;
let (base, server) = serve_router(Arc::clone(&state)).await;
let client = reqwest::Client::new();
let one: serde_json::Value = client
.get(format!("{base}/v1/boxes/{STUB_BOX_ID}"))
.send()
.await
.expect("request must reach the server")
.json()
.await
.expect("a box body");
assert_eq!(
one["auto_stop"], 900,
"GET must report the AutoStop deadline serve is sweeping, not the runtime's zero"
);
assert_eq!(
one["auto_delete"], 604_800,
"GET must report the AutoDelete deadline serve is sweeping, not the runtime's zero"
);
let listed: serde_json::Value = client
.get(format!("{base}/v1/boxes"))
.send()
.await
.expect("request must reach the server")
.json()
.await
.expect("a list body");
assert_eq!(
listed["boxes"][0]["auto_stop"], 900,
"the list must report the same deadlines the single read does"
);
assert_eq!(listed["boxes"][0]["auto_delete"], 604_800);
server.abort();
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn removing_a_box_by_name_forgets_the_deadline_filed_under_its_id() {
let upstream = stub_upstream("stopped", true).await;
let state = Arc::new(AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
state
.set_lifecycle(
STUB_BOX_ID,
LifecyclePolicy {
auto_stop: 900,
auto_delete: 604_800,
},
)
.await;
let (base, server) = serve_router(Arc::clone(&state)).await;
let response = reqwest::Client::new()
.delete(format!("{base}/v1/boxes/{STUB_BOX_NAME}"))
.send()
.await
.expect("request must reach the server");
assert_eq!(
response.status(),
StatusCode::NO_CONTENT,
"the box must actually have been removed, or the forget is untested"
);
assert_eq!(
upstream.deletes(),
1,
"the box must have been removed upstream, not just forgotten locally"
);
assert!(
!state.lifecycle.read().await.contains_key(STUB_BOX_ID),
"the deadline outlived the box it belonged to"
);
server.abort();
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_sweep_removes_a_box_whose_delete_deadline_has_passed() {
let upstream = stub_upstream("stopped", true).await;
let state = AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
};
state
.set_lifecycle(
STUB_BOX_ID,
LifecyclePolicy {
auto_stop: 0,
auto_delete: 60,
},
)
.await;
run_lifecycle_once(&state, Instant::now()).await;
assert_eq!(
upstream.deletes(),
1,
"a box at rest past its AutoDelete window must actually be removed"
);
assert!(
!state.lifecycle.read().await.contains_key(STUB_BOX_ID),
"a removed box must not keep its deadline"
);
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn creating_a_box_over_http_files_the_deadlines_it_asked_for() {
let upstream = stub_upstream("running", true).await;
let state = Arc::new(AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
let (base, server) = serve_router(Arc::clone(&state)).await;
let created: serde_json::Value = reqwest::Client::new()
.post(format!("{base}/v1/boxes"))
.json(&serde_json::json!({
"image": "alpine:latest",
"auto_stop": 900,
"auto_delete": 604_800,
}))
.send()
.await
.expect("request must reach the server")
.json()
.await
.expect("a box body");
assert_eq!(
created["auto_stop"], 900,
"the create response must echo the policy, not the runtime's zero"
);
assert_eq!(created["auto_delete"], 604_800);
assert_eq!(
state.lifecycle.read().await.get(STUB_BOX_ID).copied(),
Some(LifecyclePolicy {
auto_stop: 900,
auto_delete: 604_800,
}),
"the sweep reads this map; a policy that never lands here is never enforced"
);
server.abort();
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn activity_on_an_escaped_name_lands_under_the_name_the_sweep_reads() {
let upstream = stub_upstream("running", true).await;
let state = Arc::new(AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
let (base, server) = serve_router(Arc::clone(&state)).await;
let _ = reqwest::Client::new()
.post(format!("{base}/v1/boxes/my%20box/exec"))
.json(&serde_json::json!({"command": "echo"}))
.send()
.await
.expect("request must reach the server");
let _ = reqwest::Client::new()
.post(format!("{base}/v1/boxes/swept%2Dbox/exec"))
.json(&serde_json::json!({"command": "echo"}))
.send()
.await
.expect("request must reach the server");
let clocks = state.last_activity.read().await;
assert!(
clocks.contains_key("my box"),
"the stamp must be filed under the decoded name the sweep looks for, \
not the escaped spelling: {:?}",
clocks.keys().collect::<Vec<_>>()
);
assert!(
clocks.contains_key(STUB_BOX_NAME),
"an escaped unreserved character must decode too: {:?}",
clocks.keys().collect::<Vec<_>>()
);
drop(clocks);
server.abort();
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stopped_box_with_autoresume_off_refuses_the_routes_that_would_wake_it() {
let upstream = stub_upstream("stopped", false).await;
let state = Arc::new(AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
let (base, server) = serve_router(Arc::clone(&state)).await;
let client = reqwest::Client::new();
let exec = client
.post(format!("{base}/v1/boxes/{STUB_BOX_ID}/exec"))
.json(&serde_json::json!({"command": "echo"}))
.send()
.await
.expect("request must reach the server");
assert_eq!(
exec.status().as_u16(),
409,
"exec must not implicitly wake a box whose caller disabled AutoResume"
);
let body = exec.text().await.expect("an error body");
assert!(
body.contains("has AutoResume disabled"),
"the refusal must be the AutoResume gate, not another 409: {body}"
);
let download = client
.get(format!("{base}/v1/boxes/{STUB_BOX_ID}/files?path=/tmp"))
.send()
.await
.expect("request must reach the server");
assert_eq!(
download.status().as_u16(),
409,
"reading files must be gated too; it drives the guest just as exec does"
);
assert!(
download
.text()
.await
.expect("an error body")
.contains("has AutoResume disabled"),
"the refusal must be the AutoResume gate, not another 409"
);
let upload = client
.put(format!("{base}/v1/boxes/{STUB_BOX_ID}/files?path=/tmp/x"))
.body(Vec::new())
.send()
.await
.expect("request must reach the server");
assert_eq!(
upload.status().as_u16(),
409,
"writing files is a separate handler and resolves for itself"
);
assert!(
upload
.text()
.await
.expect("an error body")
.contains("has AutoResume disabled"),
"the refusal must be the AutoResume gate, not another 409"
);
let ws_url = base.replacen("http://", "ws://", 1);
let refused =
tokio_tungstenite::connect_async(format!("{ws_url}/v1/boxes/{STUB_BOX_ID}/attach"))
.await
.expect_err("the handshake must be refused, not completed");
match refused {
tokio_tungstenite::tungstenite::Error::Http(response) => {
assert_eq!(
response.status().as_u16(),
409,
"attach must not implicitly wake the box either"
);
let body = String::from_utf8_lossy(response.body().as_deref().unwrap_or_default())
.into_owned();
assert!(
body.contains("has AutoResume disabled"),
"the refusal must be the AutoResume gate, not the already-attached 409: {body}"
);
}
other => panic!("expected an HTTP refusal, got {other:?}"),
}
server.abort();
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stopped_box_with_autoresume_on_is_not_refused() {
let upstream = stub_upstream("stopped", true).await;
let state = Arc::new(AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
let (base, server) = serve_router(Arc::clone(&state)).await;
let exec = reqwest::Client::new()
.post(format!("{base}/v1/boxes/{STUB_BOX_ID}/exec"))
.json(&serde_json::json!({"command": "echo"}))
.send()
.await
.expect("request must reach the server");
assert_ne!(
exec.status().as_u16(),
409,
"AutoResume was enabled; the wake must not have been refused"
);
assert_eq!(
upstream.execs(),
1,
"the request must have reached the guest, not merely avoided a 409"
);
server.abort();
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_sweep_stops_a_box_idle_past_its_window() {
let upstream = stub_upstream("running", true).await;
let state = AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
};
state
.set_lifecycle(
STUB_BOX_ID,
LifecyclePolicy {
auto_stop: 60,
auto_delete: 0,
},
)
.await;
let now = Instant::now();
state
.last_activity
.write()
.await
.insert(STUB_BOX_ID.to_string(), now - Duration::from_secs(600));
run_lifecycle_once(&state, now).await;
assert_eq!(
upstream.stops(),
1,
"a running box idle past its AutoStop window must actually be stopped"
);
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_sweep_leaves_a_box_still_inside_its_idle_window() {
let upstream = stub_upstream("running", true).await;
let state = AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
};
state
.set_lifecycle(
STUB_BOX_ID,
LifecyclePolicy {
auto_stop: 600,
auto_delete: 0,
},
)
.await;
let now = Instant::now();
state
.last_activity
.write()
.await
.insert(STUB_BOX_ID.to_string(), now - Duration::from_secs(60));
run_lifecycle_once(&state, now).await;
assert_eq!(
upstream.stops(),
0,
"the sweep stopped a box that had been used inside its window"
);
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_sweep_leaves_a_box_whose_deadline_has_not_passed() {
let upstream = stub_upstream("stopped", true).await;
let state = AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
};
state
.set_lifecycle(
STUB_BOX_ID,
LifecyclePolicy {
auto_stop: 0,
auto_delete: 60 * 60 * 24 * 365 * 100,
},
)
.await;
run_lifecycle_once(&state, Instant::now()).await;
assert_eq!(
upstream.deletes(),
0,
"the sweep removed a box that was still inside its AutoDelete window"
);
assert!(
state.lifecycle.read().await.contains_key(STUB_BOX_ID),
"a box that was not removed must keep its deadline"
);
upstream.task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn scraping_metrics_will_not_wake_a_box_the_sweep_would_stop() {
let upstream = stub_upstream("stopped", true).await;
let state = Arc::new(AppState {
runtime: BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(upstream.url.clone()))
.expect("rest runtime"),
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
state
.set_lifecycle(
STUB_BOX_ID,
LifecyclePolicy {
auto_stop: 900,
auto_delete: 604_800,
},
)
.await;
let (base, server) = serve_router(Arc::clone(&state)).await;
for spelling in [STUB_BOX_ID, STUB_BOX_NAME] {
let response = reqwest::Client::new()
.get(format!("{base}/v1/boxes/{spelling}/metrics"))
.send()
.await
.expect("request must reach the server");
assert_eq!(
response.status().as_u16(),
409,
"a scrape addressed by {spelling:?} must be refused, not answered by booting the box"
);
let body = response.text().await.expect("an error body");
assert!(
body.contains("a metrics scrape does not start a box"),
"the refusal must be the metrics rule, not the AutoResume gate: {body}"
);
assert!(
!state.last_activity.read().await.contains_key(spelling),
"a scrape must not stamp the idle clock either"
);
}
server.abort();
upstream.task.abort();
}
#[tokio::test]
async fn a_rejected_request_does_not_stamp_the_idle_clock() {
let runtime = BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(
"http://127.0.0.1:1".to_string(),
))
.expect("rest runtime");
let state = Arc::new(AppState {
runtime,
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: Some("expected-key".to_string()),
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let server = {
let state = Arc::clone(&state);
tokio::spawn(async move {
let _ = axum::serve(listener, build_router(state)).await;
})
};
let unauthorized = reqwest::Client::new()
.post(format!("http://127.0.0.1:{port}/v1/boxes/victim/exec"))
.json(&serde_json::json!({"command": "echo"}))
.send()
.await
.expect("request must reach the server");
assert_eq!(unauthorized.status().as_u16(), 401);
assert!(
!state.last_activity.read().await.contains_key("victim"),
"a 401'd request must not have reset the box's idle window"
);
server.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn container_attach_route_is_registered() {
let runtime = BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(
"http://127.0.0.1:1".to_string(),
))
.expect("rest runtime");
let state = Arc::new(AppState {
runtime,
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let server = tokio::spawn(async move {
let _ = axum::serve(listener, build_router(state)).await;
});
let http = reqwest::Client::new();
let attached = http
.post(format!("http://127.0.0.1:{port}/v1/boxes/box1/attach"))
.send()
.await
.expect("POST /attach");
assert_eq!(
attached.status().as_u16(),
405,
"GET /v1/boxes/{{box_id}}/attach must be registered (405 = path matched, method did not)",
);
let withdrawn_ports = http
.get(format!("http://127.0.0.1:{port}/v1/boxes/box1/ports"))
.send()
.await
.expect("GET withdrawn /ports route");
assert_eq!(
withdrawn_ports.status().as_u16(),
404,
"the withdrawn /ports discovery route must not come back",
);
let unrouted = http
.get(format!("http://127.0.0.1:{port}/v1/boxes/box1/attach/nope"))
.send()
.await
.expect("GET unregistered path");
assert_eq!(
unrouted.status().as_u16(),
404,
"control: an unregistered path must 404, so the 405 above is meaningful",
);
server.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn late_attach_subscriber_must_see_prior_output() {
let (active, stdout_tx, _stderr_tx, _result_tx) = make_test_active();
for i in 1..=5 {
stdout_tx.send(format!("line-{i}\n")).unwrap();
}
tokio::time::sleep(Duration::from_millis(50)).await;
let mut rx = active.stdout_bus().subscribe();
stdout_tx.send("line-6\n".to_string()).unwrap();
tokio::time::sleep(Duration::from_millis(20)).await;
let mut received = Vec::new();
while let Ok(Ok(data)) = tokio::time::timeout(Duration::from_millis(50), rx.recv()).await {
received.push(String::from_utf8(data).unwrap());
}
assert!(
received.len() >= 6,
"late subscriber must see pre-subscribe output; \
got {} line(s): {:?} (expected >= 6, including the 5 pre-attach lines)",
received.len(),
received,
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn done_signal_must_wait_for_pump_completion() {
let (active, stdout_tx, stderr_tx, result_tx) = make_test_active();
let mut rx = active.stdout_bus().subscribe();
let mut done_rx = active.done_rx();
stdout_tx.send("final-line\n".to_string()).unwrap();
drop(stdout_tx);
drop(stderr_tx);
result_tx
.send(boxlite::ExecResult {
exit_code: 0,
error_message: None,
})
.unwrap();
let _ = tokio::time::timeout(Duration::from_secs(2), done_rx.changed()).await;
let mut all = Vec::new();
while let Ok(bytes) = rx.try_recv() {
all.push(String::from_utf8(bytes).unwrap());
}
assert!(
all.iter().any(|s| s.contains("final-line")),
"after done_rx fires, all output must have been broadcast; \
got: {:?}",
all,
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reaper_retains_completed_exec_for_grace_period() {
let (active, stdout_tx, stderr_tx, result_tx) = make_test_active();
drop(stdout_tx);
drop(stderr_tx);
result_tx
.send(boxlite::ExecResult {
exit_code: 42,
error_message: None,
})
.unwrap();
for _ in 0..20 {
if active.is_done() {
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(active.is_done(), "precondition: exec must be done");
let now = Instant::now();
assert!(
active.should_retain(now),
"recently-completed exec must be retained (grace period = {:?})",
COMPLETED_RETENTION_GRACE,
);
let far_future = now + COMPLETED_RETENTION_GRACE + Duration::from_secs(1);
assert!(
!active.should_retain(far_future),
"exec past the retention grace must not be retained",
);
}
#[tokio::test]
async fn find_main_session_skips_a_finished_one_so_a_restart_gets_a_new_session() {
let (exec, channels) = stub_execution("cid-main");
let finished = ActiveExecution::new("box1".to_string(), SessionKind::Main, exec, None);
let mut executions = HashMap::new();
executions.insert("cid-main".to_string(), Arc::clone(&finished));
assert!(
find_main_session(&executions, "box1").is_some(),
"precondition: a live main session is found"
);
let StubChannels(stdout_tx, stderr_tx, _stdin_rx, result_tx) = channels;
drop(stdout_tx);
drop(stderr_tx);
result_tx
.send(boxlite::ExecResult {
exit_code: 0,
error_message: None,
})
.unwrap();
for _ in 0..40 {
if finished.is_done() {
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(finished.is_done(), "precondition: the main session ended");
assert!(
find_main_session(&executions, "box1").is_none(),
"a finished main session must not be reused — the restarted box needs a new one"
);
}
#[tokio::test]
async fn reaper_never_reaps_the_boxs_main_session() {
let runtime = BoxliteRuntime::rest(boxlite::BoxliteRestOptions::new(
"http://127.0.0.1:1".to_string(),
))
.expect("rest runtime");
let state = AppState {
runtime,
boxes: RwLock::new(HashMap::new()),
executions: RwLock::new(HashMap::new()),
api_key: None,
lifecycle: RwLock::new(HashMap::new()),
last_activity: RwLock::new(HashMap::new()),
};
let (init, _init_channels) = stub_execution("cid-main");
let main = Arc::new(ActiveExecution::new(
"box1".to_string(),
SessionKind::Main,
init,
None,
));
let (tenant_exec, _tenant_channels) = stub_execution("exec-1");
let tenant = Arc::new(ActiveExecution::new(
"box1".to_string(),
SessionKind::Exec,
tenant_exec,
None,
));
{
let mut map = state.executions.write().await;
map.insert("cid-main".to_string(), Arc::clone(&main));
map.insert("exec-1".to_string(), Arc::clone(&tenant));
}
let doomsday = Instant::now() + Duration::from_secs(48 * 60 * 60);
run_reap_once(
&state,
doomsday,
Duration::from_secs(1),
Duration::from_secs(1),
Duration::from_secs(60),
)
.await;
let surviving = state.executions.read().await;
assert!(
surviving.contains_key("cid-main"),
"the box's main session must survive the reaper — reaping it kills init and destroys the box"
);
assert!(
!main.is_reaping_kill().await,
"the main session must never even be marked for kill"
);
assert!(
!surviving.contains_key("exec-1"),
"an orphaned exec past the lifetime cap must still be killed and evicted"
);
}
}