128 lines
4.2 KiB
Rust
128 lines
4.2 KiB
Rust
//! Host-side vacuum of per-agent bash-task files. The harness writes
|
|
//! three files per `bash_run` call:
|
|
//!
|
|
//! ```text
|
|
//! harness_dir()/bash-tasks/<id>.json — task metadata + status
|
|
//! harness_dir()/bash-tasks/<id>.out — captured stdout
|
|
//! harness_dir()/bash-tasks/<id>.err — captured stderr
|
|
//! ```
|
|
//!
|
|
//! Completed tasks (status `done`, `timed_out`, `interrupted`) are
|
|
//! never removed by the harness. On a busy agent they accumulate
|
|
//! indefinitely. This module sweeps every agent's `bash-tasks/`
|
|
//! directory hourly and deletes the `.json`/`.out`/`.err` trio for
|
|
//! any terminal task whose `completed_at` timestamp is older than
|
|
//! `KEEP_SECS`.
|
|
//!
|
|
//! Mirrors `events_vacuum` in structure — host-side
|
|
//! so a misbehaving harness cannot disable its own cleanup.
|
|
|
|
use std::path::Path;
|
|
use std::sync::Arc;
|
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|
|
|
use crate::coordinator::Coordinator;
|
|
|
|
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
|
|
/// Keep completed task files for 48 hours before sweeping them.
|
|
const KEEP_SECS: i64 = 48 * 3600;
|
|
|
|
/// Terminal task statuses — files for these are eligible for deletion.
|
|
const TERMINAL_STATUSES: &[&str] = &["done", "timed_out", "interrupted"];
|
|
|
|
/// Spawn the background vacuum loop as a detached tokio task.
|
|
pub fn spawn(coord: &Arc<Coordinator>) {
|
|
let mut shutdown = coord.shutdown_rx();
|
|
tokio::spawn(async move {
|
|
loop {
|
|
sweep_once();
|
|
tokio::select! {
|
|
() = tokio::time::sleep(VACUUM_INTERVAL) => {}
|
|
_ = shutdown.changed() => {
|
|
tracing::info!("bash-tasks vacuum: shutdown signal received");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
fn sweep_once() {
|
|
let cutoff = now_unix() - KEEP_SECS;
|
|
for name in Coordinator::kept_state_names() {
|
|
let tasks_dir = Coordinator::agent_harness_dir(&name).join("bash-tasks");
|
|
if !tasks_dir.is_dir() {
|
|
continue;
|
|
}
|
|
let removed = vacuum_dir(&tasks_dir, cutoff);
|
|
if removed > 0 {
|
|
tracing::info!(agent = %name, removed, "bash-tasks vacuum");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Delete eligible task trios in `dir`. Returns the count of `.json`
|
|
/// files removed (each represents one task; `.out`/`.err` deletions
|
|
/// are not counted separately).
|
|
fn vacuum_dir(dir: &Path, cutoff: i64) -> u64 {
|
|
let Ok(rd) = std::fs::read_dir(dir) else {
|
|
return 0;
|
|
};
|
|
let mut removed: u64 = 0;
|
|
for entry in rd.flatten() {
|
|
let path = entry.path();
|
|
// Only process the .json sentinel; derive sibling paths from it.
|
|
if path.extension().and_then(|e| e.to_str()) != Some("json") {
|
|
continue;
|
|
}
|
|
let Some(stem) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
|
|
continue;
|
|
};
|
|
if should_delete(&path, cutoff) {
|
|
delete_trio(dir, &stem);
|
|
removed += 1;
|
|
}
|
|
}
|
|
removed
|
|
}
|
|
|
|
/// Return `true` when the task file has a terminal status and a
|
|
/// `completed_at` older than `cutoff`.
|
|
fn should_delete(json_path: &Path, cutoff: i64) -> bool {
|
|
let Ok(raw) = std::fs::read_to_string(json_path) else {
|
|
return false;
|
|
};
|
|
let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) else {
|
|
return false;
|
|
};
|
|
let status = v.get("status").and_then(|s| s.as_str()).unwrap_or("");
|
|
if !TERMINAL_STATUSES.contains(&status) {
|
|
return false;
|
|
}
|
|
let completed_at = v
|
|
.get("completed_at")
|
|
.and_then(serde_json::Value::as_i64)
|
|
.unwrap_or(i64::MAX);
|
|
completed_at < cutoff
|
|
}
|
|
|
|
/// Delete the `.json`, `.out`, and `.err` files for a task. Errors
|
|
/// are logged but do not abort the sweep.
|
|
fn delete_trio(dir: &Path, stem: &str) {
|
|
for ext in ["json", "out", "err"] {
|
|
let path = dir.join(format!("{stem}.{ext}"));
|
|
if path.exists()
|
|
&& let Err(e) = std::fs::remove_file(&path)
|
|
{
|
|
tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed");
|
|
}
|
|
}
|
|
}
|
|
|
|
fn now_unix() -> i64 {
|
|
SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.ok()
|
|
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
.unwrap_or(0)
|
|
}
|