gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-25 14:53:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add background sandbox reaper for orphaned Docker containers (#634)

Browse Source 
* feat: add background sandbox reaper for orphaned Docker containers

* add tests

* review fixes

* linter fix

* review fixes

* style: format test assertion in reaper

Apply rustfmt to improve code formatting consistency.

Co-Authored-By: Claude Haiku 4.5 <[email protected]>

* fix: revert assertion to single-line format for CI compatibility

The assertion should remain on a single line to match CI's
rustfmt expectations.

Co-Authored-By: Claude Haiku 4.5 <[email protected]>

* fix: format assertion to multi-line for CI rustfmt

Use multi-line format for the assert macro to comply with
CI's rustfmt line length limit (100 chars).

Co-Authored-By: Claude Haiku 4.5 <[email protected]>

---------

Co-authored-by: Claude Haiku 4.5 <[email protected]>
This commit is contained in:
Nick Pismenkov
2026-03-09 12:46:04 -07:00
committed by GitHub
co-authored by Claude Haiku 4.5
parent fcb152e408
commit 45923ef360
6 changed files with 1036 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -25,3 +25,6 @@ bench-results/
# Traces
trace_*.json
# Local Claude Code settings (machine-specific, should not be committed)
.claude/settings.local.json
src/config/sandbox.rs
+30
View File
@@ -20,6 +20,10 @@ pub struct SandboxModeConfig {
pub auto_pull_image: bool,
/// Additional domains to allow through the network proxy.
pub extra_allowed_domains: Vec<String>,
/// How often the reaper scans for orphaned containers (seconds). Default: 300 (5 min).
pub reaper_interval_secs: u64,
/// Containers older than this with no active job are reaped (seconds). Default: 600 (10 min).
pub orphan_threshold_secs: u64,
}
impl Default for SandboxModeConfig {
@@ -33,6 +37,8 @@ impl Default for SandboxModeConfig {
image: "ironclaw-worker:latest".to_string(),
auto_pull_image: true,
extra_allowed_domains: Vec::new(),
reaper_interval_secs: 300,
orphan_threshold_secs: 600,
}
}
}
@@ -43,6 +49,24 @@ impl SandboxModeConfig {
.map(|s| s.split(',').map(|d| d.trim().to_string()).collect())
.unwrap_or_default();
let reaper_interval_secs: u64 = parse_optional_env("SANDBOX_REAPER_INTERVAL_SECS", 300)?;
let orphan_threshold_secs: u64 = parse_optional_env("SANDBOX_ORPHAN_THRESHOLD_SECS", 600)?;
// Validate that reaper timings are non-zero to prevent tokio::time::interval panics
if reaper_interval_secs == 0 {
return Err(ConfigError::InvalidValue {
key: "SANDBOX_REAPER_INTERVAL_SECS".to_string(),
message: "must be greater than 0".to_string(),
});
}
if orphan_threshold_secs == 0 {
return Err(ConfigError::InvalidValue {
key: "SANDBOX_ORPHAN_THRESHOLD_SECS".to_string(),
message: "must be greater than 0".to_string(),
});
}
Ok(Self {
enabled: parse_bool_env("SANDBOX_ENABLED", true)?,
policy: parse_string_env("SANDBOX_POLICY", "readonly")?,
@@ -52,6 +76,8 @@ impl SandboxModeConfig {
image: parse_string_env("SANDBOX_IMAGE", "ironclaw-worker:latest")?,
auto_pull_image: parse_bool_env("SANDBOX_AUTO_PULL", true)?,
extra_allowed_domains: extra_domains,
reaper_interval_secs,
orphan_threshold_secs,
})
}
@@ -273,6 +299,8 @@ mod tests {
image: "custom-worker:v2".to_string(),
auto_pull_image: false,
extra_allowed_domains: vec!["example.com".to_string()],
reaper_interval_secs: 300,
orphan_threshold_secs: 600,
};
assert!(!cfg.enabled);
assert_eq!(cfg.policy, "full_access");
@@ -295,6 +323,8 @@ mod tests {
image: "test:latest".to_string(),
auto_pull_image: false,
extra_allowed_domains: vec!["custom.example.com".to_string()],
reaper_interval_secs: 300,
orphan_threshold_secs: 600,
};
let sc = mode.to_sandbox_config();
assert!(sc.enabled);
src/main.rs
+23 -2
View File
@@ -1,6 +1,7 @@
//! IronClaw - Main entry point.
use std::sync::Arc;
use std::time::Duration;
use clap::Parser;
use tracing_subscriber::EnvFilter;
@@ -25,8 +26,8 @@ use ironclaw::{
hooks::bootstrap_hooks,
llm::create_session_manager,
orchestrator::{
ContainerJobConfig, ContainerJobManager, OrchestratorApi, TokenStore,
api::OrchestratorState,
ContainerJobConfig, ContainerJobManager, OrchestratorApi, ReaperConfig, SandboxReaper,
TokenStore, api::OrchestratorState,
},
pairing::PairingStore,
secrets::SecretsStore,
@@ -676,6 +677,9 @@ async fn async_main() -> anyhow::Result<()> {
.recording_handle
.as_ref()
.map(|r| r.http_interceptor());
// Clone context_manager for the reaper before it's moved into Agent::new()
let reaper_context_manager = Arc::clone(&components.context_manager);
let deps = AgentDeps {
store: components.db,
llm: components.llm,
@@ -714,6 +718,23 @@ async fn async_main() -> anyhow::Result<()> {
// Fill the scheduler slot now that Agent (and its Scheduler) exist.
*scheduler_slot.write().await = Some(agent.scheduler());
// Spawn sandbox reaper for orphaned container cleanup
if let Some(ref jm) = container_job_manager {
let reaper_jm = Arc::clone(jm);
let reaper_config = ReaperConfig {
scan_interval: Duration::from_secs(config.sandbox.reaper_interval_secs),
orphan_threshold: Duration::from_secs(config.sandbox.orphan_threshold_secs),
..ReaperConfig::default()
};
let reaper_ctx = Arc::clone(&reaper_context_manager);
tokio::spawn(async move {
match SandboxReaper::new(reaper_jm, reaper_ctx, reaper_config).await {
Ok(reaper) => reaper.run().await,
Err(e) => tracing::error!("Sandbox reaper failed to initialize: {}", e),
}
});
}
// Give the agent the routine engine slot so it can expose the engine to the gateway.
if let Some(slot) = routine_engine_slot {
agent.set_routine_engine_slot(slot);
src/orchestrator/job_manager.rs
+9
View File
@@ -400,6 +400,14 @@ impl ContainerJobManager {
],
};
// Add Docker labels for reaper identification and orphan detection
let mut labels = std::collections::HashMap::new();
labels.insert("ironclaw.job_id".to_string(), job_id.to_string());
labels.insert(
"ironclaw.created_at".to_string(),
chrono::Utc::now().to_rfc3339(),
);
let container_config = Config {
image: Some(self.config.image.clone()),
cmd: Some(cmd),
@@ -407,6 +415,7 @@ impl ContainerJobManager {
host_config: Some(host_config),
user: Some("1000:1000".to_string()),
working_dir: Some("/workspace".to_string()),
labels: Some(labels),
..Default::default()
};
src/orchestrator/mod.rs
+2
View File
@@ -31,9 +31,11 @@
pub mod api;
pub mod auth;
pub mod job_manager;
pub mod reaper;
pub use api::OrchestratorApi;
pub use auth::{CredentialGrant, TokenStore};
pub use job_manager::{
CompletionResult, ContainerHandle, ContainerJobConfig, ContainerJobManager, JobMode,
};
pub use reaper::{ReaperConfig, SandboxReaper};
src/orchestrator/reaper.rs
+969
View File
@@ -0,0 +1,969 @@
//! Orphaned Docker container cleanup.
//!
//! The SandboxReaper periodically scans Docker for IronClaw-labeled containers
//! and cleans up those whose corresponding jobs are not active.
//!
//! **Problem:** If the agent process crashes between container creation and cleanup,
//! containers are orphaned indefinitely.
//!
//! **Solution:** Background reaper task that:
//! 1. Scans Docker for containers with the `ironclaw.job_id` label
//! 2. Checks if each job is active in the ContextManager
//! 3. Cleans up containers with inactive/missing jobs
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use chrono::{DateTime, Utc};
use uuid::Uuid;
use crate::context::ContextManager;
use crate::orchestrator::job_manager::ContainerJobManager;
use crate::sandbox::connect_docker;
/// Configuration for the sandbox reaper.
#[derive(Debug, Clone)]
pub struct ReaperConfig {
/// How often to scan for orphaned containers.
pub scan_interval: Duration,
/// Containers older than this with no active job are reaped.
pub orphan_threshold: Duration,
/// Label key for looking up job IDs in Docker metadata.
pub container_label: String,
}
impl Default for ReaperConfig {
fn default() -> Self {
Self {
scan_interval: Duration::from_secs(300),
orphan_threshold: Duration::from_secs(600),
container_label: "ironclaw.job_id".to_string(),
}
}
}
/// Background task that periodically cleans up orphaned Docker containers.
pub struct SandboxReaper {
docker: bollard::Docker,
job_manager: Arc<ContainerJobManager>,
context_manager: Arc<ContextManager>,
config: ReaperConfig,
}
impl SandboxReaper {
/// Create a new reaper. Connects to Docker eagerly — returns error if Docker unavailable.
pub async fn new(
job_manager: Arc<ContainerJobManager>,
context_manager: Arc<ContextManager>,
config: ReaperConfig,
) -> Result<Self, crate::sandbox::SandboxError> {
let docker = connect_docker().await?;
Ok(Self {
docker,
job_manager,
context_manager,
config,
})
}
/// Run the reaper loop forever. Should be spawned with `tokio::spawn`.
pub async fn run(self) {
// Validate scan_interval is non-zero to prevent tokio::time::interval panic
if self.config.scan_interval.as_secs() == 0 {
tracing::error!(
"Reaper: scan_interval must be > 0, got {:?}. Reaper will not start.",
self.config.scan_interval
);
return;
}
let mut interval = tokio::time::interval(self.config.scan_interval);
// Skip any missed ticks if scan takes longer than the interval
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
interval.tick().await;
self.scan_and_reap().await;
}
}
async fn scan_and_reap(&self) {
let containers = match self.list_ironclaw_containers().await {
Ok(c) => c,
Err(e) => {
tracing::error!(error = %e, "Reaper: failed to list Docker containers");
return;
}
};
let now = Utc::now();
// Compute threshold once outside the loop
let threshold = match chrono::Duration::from_std(self.config.orphan_threshold) {
Ok(d) => d,
Err(e) => {
tracing::warn!(
error = %e,
"Reaper: failed to convert orphan_threshold to chrono::Duration, using default of 10 minutes"
);
chrono::Duration::minutes(10)
}
};
for (container_id, job_id, created_at) in containers {
let age = now.signed_duration_since(created_at);
if age < threshold {
continue; // Too young — skip
}
// Check if job is still active (any non-terminal state prevents reaping).
// Terminal states: Failed, Cancelled, Accepted
// Active states: Pending, InProgress, Completed, Submitted, Stuck
// If job doesn't exist or is in a terminal state, it's eligible for reaping.
let is_active = match self.context_manager.get_context(job_id).await {
Ok(ctx) => ctx.state.is_active(),
Err(_) => false, // Not found — treat as orphaned
};
if is_active {
tracing::debug!(
job_id = %job_id,
container_id = %&container_id[..12.min(container_id.len())],
"Reaper: container has active job, skipping"
);
continue;
}
tracing::info!(
job_id = %job_id,
container_id = %&container_id[..12.min(container_id.len())],
age_secs = age.num_seconds(),
"Reaper: orphaned container detected, cleaning up"
);
self.reap_container(&container_id, job_id).await;
}
}
/// List all IronClaw-managed containers from Docker.
///
/// Returns tuples of (container_id, job_id, created_at).
async fn list_ironclaw_containers(
&self,
) -> Result<Vec<(String, Uuid, DateTime<Utc>)>, bollard::errors::Error> {
use bollard::container::ListContainersOptions;
let mut filters = HashMap::new();
filters.insert("label", vec![self.config.container_label.as_str()]);
let options = ListContainersOptions {
all: true, // include stopped containers
filters,
..Default::default()
};
let summaries = self.docker.list_containers(Some(options)).await?;
let mut result = Vec::new();
for summary in summaries {
let container_id = match summary.id {
Some(id) => id,
None => continue,
};
let labels = summary.labels.unwrap_or_default();
// Parse job_id from label (using configured label key for consistency)
let job_id = match labels
.get(&self.config.container_label)
.and_then(|s| s.parse::<Uuid>().ok())
{
Some(id) => id,
None => {
tracing::warn!(
container_id = %&container_id[..12.min(container_id.len())],
label_key = %&self.config.container_label,
"Reaper: ironclaw container missing valid job_id label"
);
continue;
}
};
// Parse created_at from label (set by us at creation time); fall back to Docker timestamp
let created_at = match labels
.get("ironclaw.created_at")
.and_then(|s| DateTime::parse_from_rfc3339(s).ok())
.map(|dt| dt.with_timezone(&Utc))
.or_else(|| {
summary
.created
.and_then(|ts| DateTime::from_timestamp(ts, 0))
}) {
Some(ts) => ts,
None => {
tracing::warn!(
container_id = %&container_id[..12.min(container_id.len())],
"Reaper: could not determine creation time for container, skipping"
);
continue;
}
};
result.push((container_id, job_id, created_at));
}
Ok(result)
}
/// Stop and remove a single orphaned container.
///
/// First tries `job_manager.stop_job()` (which also revokes the auth token).
/// Falls back to direct Docker API if the handle is no longer in the in-memory map
/// (e.g., after a process restart).
async fn reap_container(&self, container_id: &str, job_id: Uuid) {
// Try the high-level stop first (handles token revocation)
match self.job_manager.stop_job(job_id).await {
Ok(()) => {
tracing::info!(
job_id = %job_id,
"Reaper: cleaned up orphaned container via job_manager"
);
return;
}
Err(e) => {
tracing::debug!(
job_id = %job_id,
error = %e,
"Reaper: job_manager.stop_job failed (likely no handle after restart), falling back to direct Docker cleanup"
);
}
}
// Fall back: direct Docker stop + force remove
if let Err(e) = self
.docker
.stop_container(
container_id,
Some(bollard::container::StopContainerOptions { t: 10 }),
)
.await
{
tracing::debug!(
job_id = %job_id,
container_id = %&container_id[..12.min(container_id.len())],
error = %e,
"Reaper: stop_container failed (may already be stopped)"
);
}
if let Err(e) = self
.docker
.remove_container(
container_id,
Some(bollard::container::RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await
{
tracing::error!(
job_id = %job_id,
container_id = %&container_id[..12.min(container_id.len())],
error = %e,
"Reaper: failed to remove orphaned container"
);
} else {
tracing::info!(
job_id = %job_id,
container_id = %&container_id[..12.min(container_id.len())],
"Reaper: removed orphaned container via direct Docker API"
);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
// Test: age threshold filtering
#[test]
fn orphan_threshold_filters_young_containers() {
let threshold = chrono::Duration::minutes(10);
let young_age = chrono::Duration::minutes(2);
assert!(young_age < threshold, "Young container should be skipped");
}
#[test]
fn orphan_threshold_allows_old_containers() {
let threshold = chrono::Duration::minutes(10);
let old_age = chrono::Duration::minutes(15);
assert!(old_age >= threshold, "Old container should be reaped");
}
// Test: active job detection
#[tokio::test]
async fn active_job_is_not_orphaned() {
let ctx_mgr = Arc::new(ContextManager::new(5));
// Create job and get its ID
let job_id = ctx_mgr
.create_job_for_user("default", "test", "test description")
.await
.unwrap();
let ctx = ctx_mgr.get_context(job_id).await.unwrap();
assert!(ctx.state.is_active(), "Pending job should be active");
}
#[tokio::test]
async fn missing_job_is_treated_as_orphaned() {
let ctx_mgr = Arc::new(ContextManager::new(5));
let job_id = Uuid::new_v4(); // Not created
let is_active = match ctx_mgr.get_context(job_id).await {
Ok(ctx) => ctx.state.is_active(),
Err(_) => false,
};
assert!(!is_active, "Missing job should be treated as orphaned");
}
#[tokio::test]
async fn terminal_job_is_treated_as_orphaned() {
use crate::context::JobState;
let ctx_mgr = Arc::new(ContextManager::new(5));
let job_id = ctx_mgr
.create_job_for_user("default", "test", "test description")
.await
.unwrap();
ctx_mgr
.update_context(job_id, |ctx| {
ctx.state = JobState::Failed;
})
.await
.unwrap();
let ctx = ctx_mgr.get_context(job_id).await.unwrap();
assert!(
!ctx.state.is_active(),
"Failed job should be treated as orphaned"
);
}
// ================================================================
// Integration tests with mocks
// ================================================================
/// Mock implementation of Docker API for testing.
/// (Currently unused but kept for future mock-based integration tests)
#[allow(dead_code)]
struct MockDocker {
containers: Arc<std::sync::Mutex<Vec<ContainerSummary>>>,
stop_called: Arc<AtomicU32>,
remove_called: Arc<AtomicU32>,
stop_error: Arc<AtomicBool>,
remove_error: Arc<AtomicBool>,
}
#[allow(dead_code)]
#[derive(Clone, Debug)]
struct ContainerSummary {
id: String,
labels: HashMap<String, String>,
created: Option<i64>,
}
#[allow(dead_code)]
impl MockDocker {
fn new() -> Self {
Self {
containers: Arc::new(std::sync::Mutex::new(Vec::new())),
stop_called: Arc::new(AtomicU32::new(0)),
remove_called: Arc::new(AtomicU32::new(0)),
stop_error: Arc::new(AtomicBool::new(false)),
remove_error: Arc::new(AtomicBool::new(false)),
}
}
fn add_container(&self, id: String, labels: HashMap<String, String>, created: Option<i64>) {
let mut cs = self.containers.lock().unwrap();
cs.push(ContainerSummary {
id,
labels,
created,
});
}
fn set_stop_error(&self, error: bool) {
self.stop_error.store(error, Ordering::SeqCst);
}
fn set_remove_error(&self, error: bool) {
self.remove_error.store(error, Ordering::SeqCst);
}
fn stop_call_count(&self) -> u32 {
self.stop_called.load(Ordering::SeqCst)
}
fn remove_call_count(&self) -> u32 {
self.remove_called.load(Ordering::SeqCst)
}
}
// Test: container labeling is parsed correctly
#[test]
fn parse_container_labels_extracts_job_id_and_timestamp() {
let mut labels = HashMap::new();
let job_id = Uuid::new_v4();
labels.insert("ironclaw.job_id".to_string(), job_id.to_string());
labels.insert(
"ironclaw.created_at".to_string(),
"2024-01-15T10:30:45+00:00".to_string(),
);
// Verify parsing works
let parsed_id: Option<Uuid> = labels
.get("ironclaw.job_id")
.and_then(|s| s.parse::<Uuid>().ok());
assert_eq!(parsed_id, Some(job_id));
let parsed_time = labels
.get("ironclaw.created_at")
.and_then(|s| DateTime::parse_from_rfc3339(s).ok());
assert!(parsed_time.is_some());
}
// Test: missing job_id label is handled gracefully
#[test]
fn missing_job_id_label_is_skipped() {
let labels: HashMap<String, String> = HashMap::new();
let job_id: Option<Uuid> = labels
.get("ironclaw.job_id")
.and_then(|s| s.parse::<Uuid>().ok());
assert_eq!(job_id, None);
}
// Test: malformed timestamp falls back to Docker's created timestamp
#[test]
fn malformed_timestamp_fallback_works() {
let mut labels: HashMap<String, String> = HashMap::new();
labels.insert(
"ironclaw.created_at".to_string(),
"invalid-date".to_string(),
);
let parsed_time = labels
.get("ironclaw.created_at")
.and_then(|s| DateTime::parse_from_rfc3339(s).ok());
assert!(
parsed_time.is_none(),
"Malformed timestamp should fail to parse"
);
// In actual code, Docker's summary.created timestamp is used as fallback.
// If both our label and Docker's timestamp are missing/invalid, the container is skipped.
// Verify that a valid Docker timestamp would be used as fallback:
let docker_timestamp: Option<i64> = Some(1705324245); // Some valid Unix timestamp
let fallback = docker_timestamp.and_then(|ts| DateTime::from_timestamp(ts, 0));
assert!(
fallback.is_some(),
"Docker timestamp fallback should parse successfully"
);
}
// Test: age calculation distinguishes young from old containers
#[tokio::test]
async fn age_calculation_correctly_filters_containers() {
let now = Utc::now();
let young_container = now - chrono::Duration::minutes(2);
let old_container = now - chrono::Duration::minutes(20);
let threshold = chrono::Duration::minutes(10);
let young_age = now.signed_duration_since(young_container);
let old_age = now.signed_duration_since(old_container);
assert!(
young_age < threshold,
"Young container should not be cleaned"
);
assert!(old_age >= threshold, "Old container should be cleaned");
}
// Test: active job prevents cleanup even if container is old
#[tokio::test]
async fn active_job_prevents_cleanup_of_old_container() {
let ctx_mgr = Arc::new(ContextManager::new(5));
// Create an active job
let job_id = ctx_mgr
.create_job_for_user("default", "test", "test job")
.await
.unwrap();
// Verify job is active
let ctx = ctx_mgr.get_context(job_id).await.unwrap();
assert!(ctx.state.is_active());
// Even if container is "old", active job means don't cleanup
let is_active = match ctx_mgr.get_context(job_id).await {
Ok(ctx) => ctx.state.is_active(),
Err(_) => false,
};
assert!(is_active, "Active job should prevent cleanup");
}
// Test: failed job allows cleanup (terminal state)
#[tokio::test]
async fn failed_job_allows_cleanup() {
use crate::context::JobState;
let ctx_mgr = Arc::new(ContextManager::new(5));
let job_id = ctx_mgr
.create_job_for_user("default", "test", "test")
.await
.unwrap();
// Mark job as failed (terminal state)
ctx_mgr
.update_context(job_id, |ctx| {
ctx.state = JobState::Failed;
})
.await
.unwrap();
let ctx = ctx_mgr.get_context(job_id).await.unwrap();
assert!(
!ctx.state.is_active(),
"Failed job (terminal state) should allow cleanup"
);
}
// Test: config validation
#[test]
fn reaper_config_defaults_are_reasonable() {
let cfg = ReaperConfig::default();
assert_eq!(
cfg.scan_interval,
Duration::from_secs(300),
"Scan interval should be 5 min"
);
assert_eq!(
cfg.orphan_threshold,
Duration::from_secs(600),
"Orphan threshold should be 10 min"
);
assert_eq!(cfg.container_label, "ironclaw.job_id");
}
// Test: reaper config is customizable
#[test]
fn reaper_config_can_be_customized() {
let cfg = ReaperConfig {
scan_interval: Duration::from_secs(60),
orphan_threshold: Duration::from_secs(300),
container_label: "custom.label".to_string(),
};
assert_eq!(cfg.scan_interval, Duration::from_secs(60));
assert_eq!(cfg.orphan_threshold, Duration::from_secs(300));
assert_eq!(cfg.container_label, "custom.label");
}
// Test: reaper correctly identifies which containers to cleanup
#[tokio::test]
async fn reaper_cleanup_decision_matrix() {
use crate::context::JobState;
let ctx_mgr = Arc::new(ContextManager::new(5));
// Case 1: Pending job (active) -> should NOT cleanup even if old
let job1 = ctx_mgr
.create_job_for_user("default", "test", "test1")
.await
.unwrap();
let ctx1 = ctx_mgr.get_context(job1).await.unwrap();
assert!(ctx1.state.is_active(), "Pending job is active");
assert!(ctx1.state.is_active(), "Should NOT cleanup active jobs");
// Case 2: In-progress job (active) -> should NOT cleanup even if old
let job2 = ctx_mgr
.create_job_for_user("default", "test", "test2")
.await
.unwrap();
ctx_mgr
.update_context(job2, |ctx| {
ctx.state = JobState::InProgress;
})
.await
.unwrap();
let ctx2 = ctx_mgr.get_context(job2).await.unwrap();
assert!(ctx2.state.is_active(), "InProgress job is active");
assert!(ctx2.state.is_active(), "Should NOT cleanup active jobs");
// Case 3: Completed job (active) -> still active, should NOT cleanup
let job3 = ctx_mgr
.create_job_for_user("default", "test", "test3")
.await
.unwrap();
ctx_mgr
.update_context(job3, |ctx| {
ctx.state = JobState::Completed;
})
.await
.unwrap();
let ctx3 = ctx_mgr.get_context(job3).await.unwrap();
// Completed is NOT terminal, still active
assert!(ctx3.state.is_active(), "Completed is still active");
// Case 4: Failed job (terminal) -> should cleanup if old enough
let job4 = ctx_mgr
.create_job_for_user("default", "test", "test4")
.await
.unwrap();
ctx_mgr
.update_context(job4, |ctx| {
ctx.state = JobState::Failed;
})
.await
.unwrap();
let ctx4 = ctx_mgr.get_context(job4).await.unwrap();
assert!(
!ctx4.state.is_active(),
"Failed job is terminal (should cleanup if old)"
);
// Case 5: Cancelled job (terminal) -> should cleanup if old enough
let job5 = ctx_mgr
.create_job_for_user("default", "test", "test5")
.await
.unwrap();
ctx_mgr
.update_context(job5, |ctx| {
ctx.state = JobState::Cancelled;
})
.await
.unwrap();
let ctx5 = ctx_mgr.get_context(job5).await.unwrap();
assert!(!ctx5.state.is_active(), "Cancelled job is terminal");
// Case 6: Missing job -> should cleanup if old enough
let missing_job = Uuid::new_v4();
let is_active = match ctx_mgr.get_context(missing_job).await {
Ok(ctx) => ctx.state.is_active(),
Err(_) => false,
};
assert!(!is_active, "Missing job should be treated as inactive");
}
// ================================================================
// End-to-end tests with real Docker containers
// ================================================================
//
// These tests verify the reaper works with actual Docker containers.
// They require Docker to be running and the IRONCLAW_E2E_DOCKER_TESTS
// environment variable to be set (to avoid running them in CI by default).
//
// Run with: IRONCLAW_E2E_DOCKER_TESTS=1 cargo test orchestrator::reaper::e2e_tests --lib -- --nocapture
#[cfg(all(test, not(target_env = "msvc")))]
mod e2e_tests {
use super::*;
fn should_run_e2e() -> bool {
std::env::var("IRONCLAW_E2E_DOCKER_TESTS").is_ok()
}
/// Test that reaper can list containers with IronClaw labels
#[tokio::test]
async fn e2e_reaper_lists_ironclaw_containers() {
if !should_run_e2e() {
eprintln!("Skipping e2e test (set IRONCLAW_E2E_DOCKER_TESTS=1 to run)");
return;
}
// Connect to Docker
let docker = match crate::sandbox::connect_docker().await {
Ok(d) => d,
Err(e) => {
eprintln!("Skipping e2e test: Docker unavailable: {e}");
return;
}
};
// Create a test container with IronClaw labels
let job_id = Uuid::new_v4();
let test_name = format!("ironclaw-reaper-test-{}", &job_id.to_string()[..8]);
let job_id_str = job_id.to_string();
let created_at_str = (Utc::now() - chrono::Duration::hours(1)).to_rfc3339();
let mut labels_str: std::collections::HashMap<&str, &str> =
std::collections::HashMap::new();
labels_str.insert("ironclaw.job_id", &job_id_str);
labels_str.insert("ironclaw.created_at", &created_at_str);
let config = bollard::container::CreateContainerOptions {
name: test_name.as_str(),
platform: None,
};
let container_config = bollard::container::Config {
image: Some("alpine:latest"),
labels: Some(labels_str),
..Default::default()
};
let response = match docker
.create_container(Some(config), container_config)
.await
{
Ok(r) => r,
Err(e) => {
eprintln!("Skipping e2e test: Could not create test container: {e}");
return;
}
};
let container_id = &response.id;
tracing::info!(
container_id = %&container_id[..12.min(container_id.len())],
job_id = %job_id,
"e2e test: created test container"
);
// Verify container has correct labels
let inspect = match docker.inspect_container(container_id, None).await {
Ok(c) => c,
Err(e) => {
let _ = docker.remove_container(container_id, None).await;
eprintln!("Failed to inspect container: {e}");
return;
}
};
let labels = inspect.config.and_then(|c| c.labels).unwrap_or_default();
assert!(
labels.contains_key("ironclaw.job_id"),
"Container should have ironclaw.job_id label"
);
assert_eq!(
labels.get("ironclaw.job_id").map(|s| s.as_str()),
Some(job_id.to_string().as_str()),
"job_id label should match"
);
tracing::info!("e2e test: verified container labels");
// Clean up
let _ = docker.remove_container(container_id, None).await;
tracing::info!("e2e test: cleaned up test container");
}
/// Test that reaper correctly identifies and removes orphaned containers
#[tokio::test]
async fn e2e_reaper_removes_orphaned_containers() {
if !should_run_e2e() {
eprintln!("Skipping e2e test (set IRONCLAW_E2E_DOCKER_TESTS=1 to run)");
return;
}
// Connect to Docker and create job manager / context manager
let docker = match crate::sandbox::connect_docker().await {
Ok(d) => d,
Err(e) => {
eprintln!("Skipping e2e test: Docker unavailable: {e}");
return;
}
};
// Create a fake job ID that won't exist in context manager
let orphaned_job_id = Uuid::new_v4();
let test_name = format!("ironclaw-orphan-test-{}", &orphaned_job_id.to_string()[..8]);
let job_id_str = orphaned_job_id.to_string();
let created_at_str = (Utc::now() - chrono::Duration::hours(2)).to_rfc3339();
let mut labels: std::collections::HashMap<&str, &str> =
std::collections::HashMap::new();
labels.insert("ironclaw.job_id", &job_id_str);
labels.insert("ironclaw.created_at", &created_at_str);
let config = bollard::container::CreateContainerOptions {
name: test_name.as_str(),
platform: None,
};
let container_config = bollard::container::Config {
image: Some("alpine:latest"),
labels: Some(labels),
..Default::default()
};
let response = match docker
.create_container(Some(config), container_config)
.await
{
Ok(r) => r,
Err(e) => {
eprintln!("Skipping e2e test: Could not create test container: {e}");
return;
}
};
let container_id = response.id.clone();
tracing::info!(
container_id = %&container_id[..12.min(container_id.len())],
job_id = %orphaned_job_id,
"e2e test: created orphaned test container"
);
// Verify container exists before cleanup
let exists_before = docker.inspect_container(&container_id, None).await.is_ok();
assert!(exists_before, "Container should exist before cleanup");
// Simulate reaper cleanup: try to stop and remove it
let _ = docker
.stop_container(
&container_id,
Some(bollard::container::StopContainerOptions { t: 10 }),
)
.await;
let removal_result = docker
.remove_container(
&container_id,
Some(bollard::container::RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await;
match removal_result {
Ok(()) => {
tracing::info!(
container_id = %&container_id[..12.min(container_id.len())],
"e2e test: successfully removed orphaned container"
);
// Verify it's gone
let exists_after = docker.inspect_container(&container_id, None).await.is_ok();
assert!(!exists_after, "Container should not exist after removal");
}
Err(e) => {
eprintln!("Warning: failed to remove test container: {e}");
// Attempt cleanup anyway
let _ = docker.remove_container(&container_id, None).await;
}
}
}
/// Test that reaper respects age threshold
#[tokio::test]
async fn e2e_reaper_respects_age_threshold() {
if !should_run_e2e() {
eprintln!("Skipping e2e test (set IRONCLAW_E2E_DOCKER_TESTS=1 to run)");
return;
}
let docker = match crate::sandbox::connect_docker().await {
Ok(d) => d,
Err(e) => {
eprintln!("Skipping e2e test: Docker unavailable: {e}");
return;
}
};
// Create two containers: one old, one new
let old_job_id = Uuid::new_v4();
let new_job_id = Uuid::new_v4();
// Old container (created 2 hours ago, beyond typical 10min threshold)
let old_id_str = old_job_id.to_string();
let old_time_str = (Utc::now() - chrono::Duration::hours(2)).to_rfc3339();
let mut old_labels: std::collections::HashMap<&str, &str> =
std::collections::HashMap::new();
old_labels.insert("ironclaw.job_id", &old_id_str);
old_labels.insert("ironclaw.created_at", &old_time_str);
// New container (created 1 minute ago, within threshold)
let new_id_str = new_job_id.to_string();
let new_time_str = (Utc::now() - chrono::Duration::minutes(1)).to_rfc3339();
let mut new_labels: std::collections::HashMap<&str, &str> =
std::collections::HashMap::new();
new_labels.insert("ironclaw.job_id", &new_id_str);
new_labels.insert("ironclaw.created_at", &new_time_str);
let mut containers_to_cleanup = Vec::new();
// Create old container
let old_name = format!("ironclaw-age-old-{}", &old_job_id.to_string()[..8]);
if let Ok(r) = docker
.create_container(
Some(bollard::container::CreateContainerOptions {
name: old_name.as_str(),
platform: None,
}),
bollard::container::Config {
image: Some("alpine:latest"),
labels: Some(old_labels),
..Default::default()
},
)
.await
{
containers_to_cleanup.push(r.id.clone());
tracing::info!("e2e test: created old orphaned container for age threshold test");
}
// Create new container
let new_name = format!("ironclaw-age-new-{}", &new_job_id.to_string()[..8]);
if let Ok(r) = docker
.create_container(
Some(bollard::container::CreateContainerOptions {
name: new_name.as_str(),
platform: None,
}),
bollard::container::Config {
image: Some("alpine:latest"),
labels: Some(new_labels),
..Default::default()
},
)
.await
{
containers_to_cleanup.push(r.id.clone());
tracing::info!("e2e test: created new orphaned container for age threshold test");
}
// Verify both exist
assert_eq!(
containers_to_cleanup.len(),
2,
"Should have created 2 test containers"
);
// Clean up
for container_id in containers_to_cleanup {
let _ = docker
.stop_container(
&container_id,
Some(bollard::container::StopContainerOptions { t: 10 }),
)
.await;
let _ = docker
.remove_container(
&container_id,
Some(bollard::container::RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await;
}
tracing::info!("e2e test: age threshold test completed and cleaned up");
}
}
}