Reaping: sandbox orphan reaper + DB expiry/retention sweeper + volume-init retry
Closes the cleanup gaps found in review (latent today; bites under load/crashes).
Sandbox containers (cm-sandbox / cm-runtime):
- label every sandbox `clawmates.sandbox={agent|browser}` at create
- SandboxDriver::list_managed(kind) (Docker label filter + K8s label selector)
- SandboxManager::reconcile_orphans(ttl) + spawn_reaper: removes engine
containers no live handle owns (ZERO = all)
- boot reconciliation (every pre-existing sandbox is an orphan from a dead
process) + periodic reaper (5m interval / 10m TTL)
- SIGTERM graceful drain: serve().with_graceful_shutdown → shutdown() both
managers so a redeploy can't leak; destroy errors now logged not swallowed
DB expiry/retention (cm-db cleanup.rs + cm-api cleanup_sweeper, hourly):
- expire auth_sessions + oauth_states past expires_at (security)
- prune sent outbox(7d), run_events(14d), routine_runs(30d), terminal
topology_runs(90d), consumed execution_grants(7d)
Compose: volume-init restart "no" → on-failure:5 (retry instead of wedging boot).
Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
3511c3ca10
commit
148705769f
+14
@@ -0,0 +1,14 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "DELETE FROM run_events WHERE created_at < now() - make_interval(days => $1)",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Int4"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "12dc6bffa3cb0db6c76b96f611c7885dfb5e884dbecd1fdd83e2424f6b1d7cd9"
|
||||||
|
}
|
||||||
+14
@@ -0,0 +1,14 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "DELETE FROM topology_runs\n WHERE status IN ('completed', 'failed', 'cancelled')\n AND finished_at IS NOT NULL\n AND finished_at < now() - make_interval(days => $1)",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Int4"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "150444818f3a8542a2fa3904d790137d1c5d2fd8720aedce51a86b32cc0f97ce"
|
||||||
|
}
|
||||||
+14
@@ -0,0 +1,14 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "DELETE FROM routine_runs WHERE started_at < now() - make_interval(days => $1)",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Int4"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "52fd56559b475ff8cf9a988d3d135163cdcec8e44b79e0d11641de6243228ce2"
|
||||||
|
}
|
||||||
+14
@@ -0,0 +1,14 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "DELETE FROM outbox WHERE status = 'sent' AND created_at < now() - make_interval(days => $1)",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Int4"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "a5859cee44f02a25eaf275b7a7bc8bd5dfc0b32dd859f052c75cc7917d96abbd"
|
||||||
|
}
|
||||||
+14
@@ -0,0 +1,14 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "DELETE FROM execution_grants\n WHERE consumed = true AND consumed_at < now() - make_interval(days => $1)",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Int4"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "d0ad9d5aa703a65927a4e60788d32053dfbef96cb789d1f99177fa05faa1a673"
|
||||||
|
}
|
||||||
+12
@@ -0,0 +1,12 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "DELETE FROM auth_sessions WHERE expires_at < now()",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": []
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "da095f50f527dd650bbf3cf76936366f03865a8b0d456d1d85ad8c56cc1ab7cc"
|
||||||
|
}
|
||||||
+12
@@ -0,0 +1,12 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "DELETE FROM oauth_states WHERE expires_at < now()",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": []
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "ef98cf6a1d204a79a983cddd2771a72a0f5a8159603b122dad90468280f32084"
|
||||||
|
}
|
||||||
@@ -22,7 +22,7 @@ cm-sandbox = { path = "../../cm-sandbox" }
|
|||||||
cm-scheduler = { path = "../../cm-scheduler" }
|
cm-scheduler = { path = "../../cm-scheduler" }
|
||||||
cm-domain = { path = "../../cm-domain" }
|
cm-domain = { path = "../../cm-domain" }
|
||||||
time = { workspace = true }
|
time = { workspace = true }
|
||||||
tokio = { workspace = true }
|
tokio = { workspace = true, features = ["signal"] }
|
||||||
|
|
||||||
[lints]
|
[lints]
|
||||||
workspace = true
|
workspace = true
|
||||||
|
|||||||
@@ -155,18 +155,31 @@ async fn run() -> Result<(), String> {
|
|||||||
driver.clone(),
|
driver.clone(),
|
||||||
&config.sandbox.image,
|
&config.sandbox.image,
|
||||||
));
|
));
|
||||||
|
let browser = std::sync::Arc::new(
|
||||||
|
cm_runtime::SandboxManager::new(driver, &config.sandbox.browser_image)
|
||||||
|
.with_egress(),
|
||||||
|
);
|
||||||
|
// Boot reconciliation: any sandbox the engine still holds is an
|
||||||
|
// orphan from a dead process (we track none yet) — remove them
|
||||||
|
// before warming so a crash/redeploy can't leak containers.
|
||||||
|
agents.reconcile_orphans(std::time::Duration::ZERO).await;
|
||||||
|
browser.reconcile_orphans(std::time::Duration::ZERO).await;
|
||||||
let agents = if config.sandbox.warm_pool > 0 {
|
let agents = if config.sandbox.warm_pool > 0 {
|
||||||
agents.warm(config.sandbox.warm_pool)
|
agents.warm(config.sandbox.warm_pool)
|
||||||
} else {
|
} else {
|
||||||
agents
|
agents
|
||||||
};
|
};
|
||||||
(
|
// Periodic reaper: catch leaks while we run (idle > 10 min and
|
||||||
Some(agents),
|
// owned by no live handle), checked every 5 min.
|
||||||
Some(std::sync::Arc::new(
|
agents.clone().spawn_reaper(
|
||||||
cm_runtime::SandboxManager::new(driver, &config.sandbox.browser_image)
|
std::time::Duration::from_secs(300),
|
||||||
.with_egress(),
|
std::time::Duration::from_secs(600),
|
||||||
)),
|
);
|
||||||
)
|
browser.clone().spawn_reaper(
|
||||||
|
std::time::Duration::from_secs(300),
|
||||||
|
std::time::Duration::from_secs(600),
|
||||||
|
);
|
||||||
|
(Some(agents), Some(browser))
|
||||||
}
|
}
|
||||||
Err(error) => {
|
Err(error) => {
|
||||||
eprintln!("clawmates-server: sandbox engine unavailable: {error}");
|
eprintln!("clawmates-server: sandbox engine unavailable: {error}");
|
||||||
@@ -176,6 +189,10 @@ async fn run() -> Result<(), String> {
|
|||||||
} else {
|
} else {
|
||||||
(None, None)
|
(None, None)
|
||||||
};
|
};
|
||||||
|
// Keep handles for the graceful-shutdown drain (SIGTERM) — the managers
|
||||||
|
// themselves are moved into the runtime config below.
|
||||||
|
let drain_agents = sandboxes.clone();
|
||||||
|
let drain_browser = browser.clone();
|
||||||
let runtime = Runtime::with_blob_store(
|
let runtime = Runtime::with_blob_store(
|
||||||
pool.clone(),
|
pool.clone(),
|
||||||
provider,
|
provider,
|
||||||
@@ -203,6 +220,9 @@ async fn run() -> Result<(), String> {
|
|||||||
// Outbound-email delivery: drains the §15-gated `outbox` over SMTP. Inert
|
// Outbound-email delivery: drains the §15-gated `outbox` over SMTP. Inert
|
||||||
// until CLAWMATES_SMTP_* is set, so it ships safely before credentials exist.
|
// until CLAWMATES_SMTP_* is set, so it ships safely before credentials exist.
|
||||||
cm_runtime::spawn_drainer(pool.clone(), std::time::Duration::from_secs(10));
|
cm_runtime::spawn_drainer(pool.clone(), std::time::Duration::from_secs(10));
|
||||||
|
// Expiry/retention sweep: expires stale auth/oauth rows and prunes old
|
||||||
|
// journal/audit rows hourly so unbounded tables don't accumulate.
|
||||||
|
cm_api::cleanup_sweeper::spawn(pool.clone(), std::time::Duration::from_secs(3600));
|
||||||
|
|
||||||
// Hosted identity (Clerk / OIDC): pin the issuer and load its JWKS.
|
// Hosted identity (Clerk / OIDC): pin the issuer and load its JWKS.
|
||||||
let auth_verifier = match config.auth.mode {
|
let auth_verifier = match config.auth.mode {
|
||||||
@@ -235,7 +255,27 @@ async fn run() -> Result<(), String> {
|
|||||||
.await
|
.await
|
||||||
.map_err(|e| format!("bind {} failed: {e}", config.listen_addr))?;
|
.map_err(|e| format!("bind {} failed: {e}", config.listen_addr))?;
|
||||||
println!("clawmates-server listening on {}", config.listen_addr);
|
println!("clawmates-server listening on {}", config.listen_addr);
|
||||||
|
// Graceful shutdown: on SIGTERM/Ctrl-C, stop accepting, finish in-flight
|
||||||
|
// requests, then DRAIN the sandbox managers so no container is left running.
|
||||||
|
let shutdown = async move {
|
||||||
|
let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
|
||||||
|
.expect("install SIGTERM handler");
|
||||||
|
tokio::select! {
|
||||||
|
_ = term.recv() => {}
|
||||||
|
_ = tokio::signal::ctrl_c() => {}
|
||||||
|
}
|
||||||
|
eprintln!("clawmates-server: shutdown signal received");
|
||||||
|
};
|
||||||
axum::serve(listener, app)
|
axum::serve(listener, app)
|
||||||
|
.with_graceful_shutdown(shutdown)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| format!("server error: {e}"))
|
.map_err(|e| format!("server error: {e}"))?;
|
||||||
|
eprintln!("clawmates-server: draining sandboxes…");
|
||||||
|
if let Some(agents) = drain_agents {
|
||||||
|
agents.shutdown().await;
|
||||||
|
}
|
||||||
|
if let Some(browser) = drain_browser {
|
||||||
|
browser.shutdown().await;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
//! Background expiry/retention sweeper. Mirrors the durable-run worker: a single
|
||||||
|
//! periodic loop that deletes overdue/stale rows so security-relevant tables
|
||||||
|
//! (auth sessions, oauth states) don't leak credentials and journal/audit tables
|
||||||
|
//! don't grow unbounded. Best-effort — a failed sweep is logged and retried next
|
||||||
|
//! tick. The first tick fires immediately on boot.
|
||||||
|
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use sqlx::PgPool;
|
||||||
|
|
||||||
|
use cm_db::repo::cleanup;
|
||||||
|
|
||||||
|
// Retention windows (days). Security-relevant rows expire with no grace.
|
||||||
|
const OUTBOX_SENT_DAYS: i32 = 7;
|
||||||
|
const RUN_EVENTS_DAYS: i32 = 14; // beyond the live SSE-replay window
|
||||||
|
const ROUTINE_RUNS_DAYS: i32 = 30;
|
||||||
|
const TOPOLOGY_RUNS_DAYS: i32 = 90; // user-visible run history
|
||||||
|
const CONSUMED_GRANTS_DAYS: i32 = 7;
|
||||||
|
|
||||||
|
/// Spawn the cleanup sweeper, running every `interval` (e.g. hourly).
|
||||||
|
pub fn spawn(pool: PgPool, interval: Duration) {
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let mut tick = tokio::time::interval(interval);
|
||||||
|
loop {
|
||||||
|
tick.tick().await;
|
||||||
|
sweep_once(&pool).await;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn sweep_once(pool: &PgPool) {
|
||||||
|
let steps: [(&str, Result<u64, cm_db::DbError>); 7] = [
|
||||||
|
("auth_sessions", cleanup::expire_auth_sessions(pool).await),
|
||||||
|
("oauth_states", cleanup::expire_oauth_states(pool).await),
|
||||||
|
(
|
||||||
|
"outbox",
|
||||||
|
cleanup::prune_sent_outbox(pool, OUTBOX_SENT_DAYS).await,
|
||||||
|
),
|
||||||
|
(
|
||||||
|
"run_events",
|
||||||
|
cleanup::prune_run_events(pool, RUN_EVENTS_DAYS).await,
|
||||||
|
),
|
||||||
|
(
|
||||||
|
"routine_runs",
|
||||||
|
cleanup::prune_routine_runs(pool, ROUTINE_RUNS_DAYS).await,
|
||||||
|
),
|
||||||
|
(
|
||||||
|
"topology_runs",
|
||||||
|
cleanup::prune_topology_runs(pool, TOPOLOGY_RUNS_DAYS).await,
|
||||||
|
),
|
||||||
|
(
|
||||||
|
"execution_grants",
|
||||||
|
cleanup::prune_consumed_grants(pool, CONSUMED_GRANTS_DAYS).await,
|
||||||
|
),
|
||||||
|
];
|
||||||
|
let mut total = 0u64;
|
||||||
|
for (table, result) in steps {
|
||||||
|
match result {
|
||||||
|
Ok(n) => total += n,
|
||||||
|
Err(e) => eprintln!("cleanup_sweeper: {table} sweep failed: {e}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if total > 0 {
|
||||||
|
eprintln!("cleanup_sweeper: removed {total} stale row(s)");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,5 +1,6 @@
|
|||||||
//! REST API for Clawmates (spec §13). One route resource per module.
|
//! REST API for Clawmates (spec §13). One route resource per module.
|
||||||
|
|
||||||
|
pub mod cleanup_sweeper;
|
||||||
mod error;
|
mod error;
|
||||||
mod extract;
|
mod extract;
|
||||||
mod mcp_door;
|
mod mcp_door;
|
||||||
|
|||||||
@@ -0,0 +1,90 @@
|
|||||||
|
//! Expiry + retention sweeps for tables that would otherwise grow unbounded.
|
||||||
|
//! Security-relevant rows (expired auth sessions, stale oauth states) are
|
||||||
|
//! deleted as soon as they're overdue; audit/journal tables get a generous
|
||||||
|
//! retention window. Each fn returns the number of rows removed.
|
||||||
|
//!
|
||||||
|
//! This is the row-level analogue of the durable-run `requeue_stale` sweep and
|
||||||
|
//! the sandbox reaper — driven on an interval by `cm_api::cleanup_sweeper`.
|
||||||
|
|
||||||
|
use sqlx::PgPool;
|
||||||
|
|
||||||
|
use crate::DbError;
|
||||||
|
|
||||||
|
/// Delete auth sessions past their `expires_at` (a revoked/expired bearer token
|
||||||
|
/// must not linger). Security-relevant — no grace window.
|
||||||
|
pub async fn expire_auth_sessions(pool: &PgPool) -> Result<u64, DbError> {
|
||||||
|
let r = sqlx::query!("DELETE FROM auth_sessions WHERE expires_at < now()")
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(r.rows_affected())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Delete OAuth CSRF states past their `expires_at` (abandoned flows).
|
||||||
|
pub async fn expire_oauth_states(pool: &PgPool) -> Result<u64, DbError> {
|
||||||
|
let r = sqlx::query!("DELETE FROM oauth_states WHERE expires_at < now()")
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(r.rows_affected())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prune successfully-sent outbox rows older than `days` (failed rows are kept
|
||||||
|
/// for inspection).
|
||||||
|
pub async fn prune_sent_outbox(pool: &PgPool, days: i32) -> Result<u64, DbError> {
|
||||||
|
let r = sqlx::query!(
|
||||||
|
"DELETE FROM outbox WHERE status = 'sent' AND created_at < now() - make_interval(days => $1)",
|
||||||
|
days,
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(r.rows_affected())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prune the SSE run-event journal older than `days` (beyond the live-replay
|
||||||
|
/// window). Removes events for long-finished runs.
|
||||||
|
pub async fn prune_run_events(pool: &PgPool, days: i32) -> Result<u64, DbError> {
|
||||||
|
let r = sqlx::query!(
|
||||||
|
"DELETE FROM run_events WHERE created_at < now() - make_interval(days => $1)",
|
||||||
|
days,
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(r.rows_affected())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prune routine-run history older than `days`.
|
||||||
|
pub async fn prune_routine_runs(pool: &PgPool, days: i32) -> Result<u64, DbError> {
|
||||||
|
let r = sqlx::query!(
|
||||||
|
"DELETE FROM routine_runs WHERE started_at < now() - make_interval(days => $1)",
|
||||||
|
days,
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(r.rows_affected())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prune terminal topology runs older than `days` (user-visible run history —
|
||||||
|
/// keep a generous window).
|
||||||
|
pub async fn prune_topology_runs(pool: &PgPool, days: i32) -> Result<u64, DbError> {
|
||||||
|
let r = sqlx::query!(
|
||||||
|
"DELETE FROM topology_runs
|
||||||
|
WHERE status IN ('completed', 'failed', 'cancelled')
|
||||||
|
AND finished_at IS NOT NULL
|
||||||
|
AND finished_at < now() - make_interval(days => $1)",
|
||||||
|
days,
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(r.rows_affected())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prune consumed single-use execution grants older than `days`.
|
||||||
|
pub async fn prune_consumed_grants(pool: &PgPool, days: i32) -> Result<u64, DbError> {
|
||||||
|
let r = sqlx::query!(
|
||||||
|
"DELETE FROM execution_grants
|
||||||
|
WHERE consumed = true AND consumed_at < now() - make_interval(days => $1)",
|
||||||
|
days,
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(r.rows_affected())
|
||||||
|
}
|
||||||
@@ -1,5 +1,6 @@
|
|||||||
pub mod agents;
|
pub mod agents;
|
||||||
pub mod audit;
|
pub mod audit;
|
||||||
|
pub mod cleanup;
|
||||||
pub mod companies;
|
pub mod companies;
|
||||||
pub mod connections;
|
pub mod connections;
|
||||||
pub mod credits;
|
pub mod credits;
|
||||||
|
|||||||
@@ -11,6 +11,8 @@ mod tools;
|
|||||||
|
|
||||||
pub use events::{RunEventBody, RunEventEnvelope};
|
pub use events::{RunEventBody, RunEventEnvelope};
|
||||||
pub use outbox::{drain_once, spawn_drainer, EmailSender, LettreSender, SmtpConfig};
|
pub use outbox::{drain_once, spawn_drainer, EmailSender, LettreSender, SmtpConfig};
|
||||||
pub use runtime::{judge_model, ProviderRegistry, Runtime, RuntimeConfig, RuntimeError, StartedRun};
|
pub use runtime::{
|
||||||
|
judge_model, ProviderRegistry, Runtime, RuntimeConfig, RuntimeError, StartedRun,
|
||||||
|
};
|
||||||
pub use sandboxes::SandboxManager;
|
pub use sandboxes::SandboxManager;
|
||||||
pub use tools::{ClockNow, EmailSend, Tool, ToolContext, ToolRegistry};
|
pub use tools::{ClockNow, EmailSend, Tool, ToolContext, ToolRegistry};
|
||||||
|
|||||||
@@ -65,9 +65,8 @@ impl LettreSender {
|
|||||||
cfg.user.clone(),
|
cfg.user.clone(),
|
||||||
cfg.pass.clone(),
|
cfg.pass.clone(),
|
||||||
);
|
);
|
||||||
let transport = lettre::AsyncSmtpTransport::<lettre::Tokio1Executor>::starttls_relay(
|
let transport =
|
||||||
&cfg.host,
|
lettre::AsyncSmtpTransport::<lettre::Tokio1Executor>::starttls_relay(&cfg.host)
|
||||||
)
|
|
||||||
.map_err(|e| e.to_string())?
|
.map_err(|e| e.to_string())?
|
||||||
.port(cfg.port)
|
.port(cfg.port)
|
||||||
.credentials(creds)
|
.credentials(creds)
|
||||||
|
|||||||
@@ -4,6 +4,7 @@
|
|||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use cm_domain::AgentId;
|
use cm_domain::AgentId;
|
||||||
use cm_sandbox::{ExecResult, SandboxDriver, SandboxHandle, SandboxSpec};
|
use cm_sandbox::{ExecResult, SandboxDriver, SandboxHandle, SandboxSpec};
|
||||||
@@ -129,18 +130,86 @@ impl SandboxManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Destroys every sandbox this manager provisioned — assigned and
|
/// Destroys every sandbox this manager provisioned — assigned and
|
||||||
/// pooled — and stops the warmer.
|
/// pooled — and stops the warmer. Called on graceful shutdown (SIGTERM).
|
||||||
pub async fn shutdown(&self) {
|
pub async fn shutdown(&self) {
|
||||||
self.warm_target
|
self.warm_target
|
||||||
.store(0, std::sync::atomic::Ordering::Relaxed);
|
.store(0, std::sync::atomic::Ordering::Relaxed);
|
||||||
let mut handles = self.handles.lock().await;
|
let mut handles = self.handles.lock().await;
|
||||||
for (_, handle) in handles.drain() {
|
for (_, handle) in handles.drain() {
|
||||||
let _ = self.driver.destroy(&handle).await;
|
if let Err(e) = self.driver.destroy(&handle).await {
|
||||||
|
eprintln!("sandbox shutdown: failed to remove {}: {e}", handle.id);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
drop(handles);
|
drop(handles);
|
||||||
let mut pool = self.pool.lock().await;
|
let mut pool = self.pool.lock().await;
|
||||||
for handle in pool.drain(..) {
|
for handle in pool.drain(..) {
|
||||||
let _ = self.driver.destroy(&handle).await;
|
if let Err(e) = self.driver.destroy(&handle).await {
|
||||||
|
eprintln!("sandbox shutdown: failed to remove {}: {e}", handle.id);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Reap orphaned sandboxes of this manager's kind: containers the engine
|
||||||
|
/// still holds that no live handle owns and that are older than
|
||||||
|
/// `older_than` (ZERO = remove all). Returns how many were removed. Run at
|
||||||
|
/// boot with ZERO (every pre-existing sandbox is an orphan from a dead
|
||||||
|
/// process) and periodically with a TTL (catches leaks while we run).
|
||||||
|
pub async fn reconcile_orphans(&self, older_than: Duration) -> usize {
|
||||||
|
let kind = cm_sandbox::sandbox_kind(self.egress);
|
||||||
|
let managed = match self.driver.list_managed(kind).await {
|
||||||
|
Ok(m) => m,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("sandbox reaper: list({kind}) failed: {e}");
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let live: std::collections::HashSet<String> = {
|
||||||
|
let handles = self.handles.lock().await;
|
||||||
|
let pool = self.pool.lock().await;
|
||||||
|
handles
|
||||||
|
.values()
|
||||||
|
.map(|h| h.id.clone())
|
||||||
|
.chain(pool.iter().map(|h| h.id.clone()))
|
||||||
|
.collect()
|
||||||
|
};
|
||||||
|
let now = std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.map(|d| d.as_secs() as i64)
|
||||||
|
.unwrap_or(0);
|
||||||
|
let min_age = older_than.as_secs() as i64;
|
||||||
|
let mut reaped = 0;
|
||||||
|
for m in managed {
|
||||||
|
if live.contains(&m.id) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if min_age > 0 && now - m.created_unix < min_age {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let handle = SandboxHandle {
|
||||||
|
id: m.id.clone(),
|
||||||
|
name: m.id.clone(),
|
||||||
|
};
|
||||||
|
match self.driver.destroy(&handle).await {
|
||||||
|
Ok(()) => reaped += 1,
|
||||||
|
Err(e) => eprintln!("sandbox reaper: failed to remove {}: {e}", m.id),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
reaped
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Background reaper: every `interval`, remove orphaned sandboxes older than
|
||||||
|
/// `ttl` that no live handle owns (covers a leak while the process stays up,
|
||||||
|
/// e.g. a panic between provision and assignment).
|
||||||
|
pub fn spawn_reaper(self: Arc<Self>, interval: Duration, ttl: Duration) {
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let mut tick = tokio::time::interval(interval);
|
||||||
|
loop {
|
||||||
|
tick.tick().await;
|
||||||
|
let n = self.reconcile_orphans(ttl).await;
|
||||||
|
if n > 0 {
|
||||||
|
eprintln!("sandbox reaper: removed {n} orphan sandbox(es)");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ use bollard::query_parameters::{
|
|||||||
use bollard::Docker;
|
use bollard::Docker;
|
||||||
use futures::StreamExt;
|
use futures::StreamExt;
|
||||||
|
|
||||||
use crate::spec::{ExecResult, SandboxHandle, SandboxSpec};
|
use crate::spec::{ExecResult, ManagedSandbox, SandboxHandle, SandboxSpec};
|
||||||
use crate::{SandboxDriver, SandboxError};
|
use crate::{SandboxDriver, SandboxError};
|
||||||
|
|
||||||
/// The seccomp deny profile, embedded so the Docker path needs no file
|
/// The seccomp deny profile, embedded so the Docker path needs no file
|
||||||
@@ -100,6 +100,11 @@ impl SandboxDriver for DockerDriver {
|
|||||||
user: Some("10001:10001".into()),
|
user: Some("10001:10001".into()),
|
||||||
cmd: Some(vec!["sleep".into(), "infinity".into()]),
|
cmd: Some(vec!["sleep".into(), "infinity".into()]),
|
||||||
host_config: Some(host_config),
|
host_config: Some(host_config),
|
||||||
|
// Mark every sandbox so the reaper can find orphans after a crash.
|
||||||
|
labels: Some(std::collections::HashMap::from([(
|
||||||
|
crate::SANDBOX_LABEL.to_string(),
|
||||||
|
crate::sandbox_kind(spec.egress).to_string(),
|
||||||
|
)])),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -192,4 +197,32 @@ impl SandboxDriver for DockerDriver {
|
|||||||
Err(e) => Err(SandboxError::Engine(e.to_string())),
|
Err(e) => Err(SandboxError::Engine(e.to_string())),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn list_managed(&self, kind: &str) -> Result<Vec<ManagedSandbox>, SandboxError> {
|
||||||
|
use bollard::query_parameters::ListContainersOptions;
|
||||||
|
let mut filters = std::collections::HashMap::new();
|
||||||
|
filters.insert(
|
||||||
|
"label".to_string(),
|
||||||
|
vec![format!("{}={}", crate::SANDBOX_LABEL, kind)],
|
||||||
|
);
|
||||||
|
let opts = ListContainersOptions {
|
||||||
|
all: true, // include stopped/exited orphans too
|
||||||
|
filters: Some(filters),
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
let list = self
|
||||||
|
.docker
|
||||||
|
.list_containers(Some(opts))
|
||||||
|
.await
|
||||||
|
.map_err(|e| SandboxError::Engine(e.to_string()))?;
|
||||||
|
Ok(list
|
||||||
|
.into_iter()
|
||||||
|
.filter_map(|c| {
|
||||||
|
c.id.map(|id| ManagedSandbox {
|
||||||
|
id,
|
||||||
|
created_unix: c.created.unwrap_or(0),
|
||||||
|
})
|
||||||
|
})
|
||||||
|
.collect())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ use kube::Api;
|
|||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
use tokio::io::AsyncReadExt;
|
use tokio::io::AsyncReadExt;
|
||||||
|
|
||||||
use crate::spec::{ExecResult, SandboxHandle, SandboxSpec};
|
use crate::spec::{ExecResult, ManagedSandbox, SandboxHandle, SandboxSpec};
|
||||||
use crate::{SandboxDriver, SandboxError};
|
use crate::{SandboxDriver, SandboxError};
|
||||||
|
|
||||||
fn engine_err(e: impl std::fmt::Display) -> SandboxError {
|
fn engine_err(e: impl std::fmt::Display) -> SandboxError {
|
||||||
@@ -145,7 +145,10 @@ impl K8sDriver {
|
|||||||
"metadata": {
|
"metadata": {
|
||||||
"name": spec.name,
|
"name": spec.name,
|
||||||
"namespace": self.namespace,
|
"namespace": self.namespace,
|
||||||
"labels": { "app.kubernetes.io/name": "clawmates-sandbox" }
|
"labels": {
|
||||||
|
"app.kubernetes.io/name": "clawmates-sandbox",
|
||||||
|
crate::SANDBOX_LABEL: crate::sandbox_kind(spec.egress)
|
||||||
|
}
|
||||||
},
|
},
|
||||||
"spec": {
|
"spec": {
|
||||||
"restartPolicy": "Never",
|
"restartPolicy": "Never",
|
||||||
@@ -282,4 +285,25 @@ impl SandboxDriver for K8sDriver {
|
|||||||
None => Ok(false),
|
None => Ok(false),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn list_managed(&self, kind: &str) -> Result<Vec<ManagedSandbox>, SandboxError> {
|
||||||
|
let lp =
|
||||||
|
kube::api::ListParams::default().labels(&format!("{}={}", crate::SANDBOX_LABEL, kind));
|
||||||
|
let list = self.pods().list(&lp).await.map_err(engine_err)?;
|
||||||
|
Ok(list
|
||||||
|
.into_iter()
|
||||||
|
.filter_map(|p| {
|
||||||
|
let name = p.metadata.name?;
|
||||||
|
let created = p
|
||||||
|
.metadata
|
||||||
|
.creation_timestamp
|
||||||
|
.map(|t| t.0.timestamp())
|
||||||
|
.unwrap_or(0);
|
||||||
|
Some(ManagedSandbox {
|
||||||
|
id: name,
|
||||||
|
created_unix: created,
|
||||||
|
})
|
||||||
|
})
|
||||||
|
.collect())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,7 +12,20 @@ mod spec;
|
|||||||
pub use docker::DockerDriver;
|
pub use docker::DockerDriver;
|
||||||
#[cfg(feature = "k8s")]
|
#[cfg(feature = "k8s")]
|
||||||
pub use k8s::K8sDriver;
|
pub use k8s::K8sDriver;
|
||||||
pub use spec::{ExecResult, SandboxHandle, SandboxSpec};
|
pub use spec::{ExecResult, ManagedSandbox, SandboxHandle, SandboxSpec};
|
||||||
|
|
||||||
|
/// Label every Clawmates sandbox carries, so orphans can be found + reaped
|
||||||
|
/// after a crash/restart. Value is the kind: `agent` (no egress) or `browser`.
|
||||||
|
pub const SANDBOX_LABEL: &str = "clawmates.sandbox";
|
||||||
|
|
||||||
|
/// The label value for a sandbox of the given kind.
|
||||||
|
pub fn sandbox_kind(egress: bool) -> &'static str {
|
||||||
|
if egress {
|
||||||
|
"browser"
|
||||||
|
} else {
|
||||||
|
"agent"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug, thiserror::Error)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
pub enum SandboxError {
|
pub enum SandboxError {
|
||||||
@@ -33,4 +46,8 @@ pub trait SandboxDriver: Send + Sync {
|
|||||||
async fn destroy(&self, handle: &SandboxHandle) -> Result<(), SandboxError>;
|
async fn destroy(&self, handle: &SandboxHandle) -> Result<(), SandboxError>;
|
||||||
/// Whether the sandbox container is currently running.
|
/// Whether the sandbox container is currently running.
|
||||||
async fn health(&self, handle: &SandboxHandle) -> Result<bool, SandboxError>;
|
async fn health(&self, handle: &SandboxHandle) -> Result<bool, SandboxError>;
|
||||||
|
/// List sandboxes the engine currently holds for this kind
|
||||||
|
/// (`agent`/`browser`), so the manager can reap orphans whose owning
|
||||||
|
/// process died. Filtered by the [`SANDBOX_LABEL`] label.
|
||||||
|
async fn list_managed(&self, kind: &str) -> Result<Vec<ManagedSandbox>, SandboxError>;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -30,3 +30,13 @@ pub struct ExecResult {
|
|||||||
pub stdout: String,
|
pub stdout: String,
|
||||||
pub stderr: String,
|
pub stderr: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A sandbox the driver currently knows about (label-filtered), used by the
|
||||||
|
/// reaper to find orphans — containers that outlived the process that made them.
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct ManagedSandbox {
|
||||||
|
/// Engine container/pod id.
|
||||||
|
pub id: String,
|
||||||
|
/// Creation time, unix seconds (for TTL-based reaping).
|
||||||
|
pub created_unix: i64,
|
||||||
|
}
|
||||||
|
|||||||
@@ -47,7 +47,8 @@ services:
|
|||||||
|
|
||||||
# Named volumes mount root-owned, but the broker runs as uid 10001 from
|
# Named volumes mount root-owned, but the broker runs as uid 10001 from
|
||||||
# a scratch image (no shell to chown itself). This one-shot prepares the
|
# a scratch image (no shell to chown itself). This one-shot prepares the
|
||||||
# socket + key volumes, then exits.
|
# socket + key volumes, then exits. `on-failure` so a transient chown error
|
||||||
|
# retries instead of wedging boot (broker/server depend on it completing).
|
||||||
volume-init:
|
volume-init:
|
||||||
image: busybox:1.36
|
image: busybox:1.36
|
||||||
command: ["sh", "-c", "chown -R 10001:10001 /run/clawmates /etc/clawmates-broker"]
|
command: ["sh", "-c", "chown -R 10001:10001 /run/clawmates /etc/clawmates-broker"]
|
||||||
@@ -55,7 +56,7 @@ services:
|
|||||||
volumes:
|
volumes:
|
||||||
- broker_run:/run/clawmates
|
- broker_run:/run/clawmates
|
||||||
- broker_key:/etc/clawmates-broker
|
- broker_key:/etc/clawmates-broker
|
||||||
restart: "no"
|
restart: on-failure:5
|
||||||
|
|
||||||
# The secret broker: separate process, separate image; credentials
|
# The secret broker: separate process, separate image; credentials
|
||||||
# never leave it. Reachable only via the shared unix socket volume.
|
# never leave it. Reachable only via the shared unix socket volume.
|
||||||
|
|||||||
Reference in New Issue
Block a user