use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RunRecord {
pub id: String,
pub pid: u32,
pub task: String,
pub status: String,
pub started_unix: i64,
pub updated_unix: i64,
}
impl RunRecord {
pub fn effective_status(&self) -> &str {
if self.status == "Running" && !pid_alive(self.pid) {
"Stale"
} else {
self.status.as_str()
}
}
}
pub fn pid_alive(pid: u32) -> bool {
use sysinfo::{Pid, ProcessesToUpdate, System};
let target = Pid::from_u32(pid);
let mut sys = System::new();
sys.refresh_processes(ProcessesToUpdate::Some(&[target]), true);
sys.process(target).is_some()
}
pub fn is_terminal_status(status: &str) -> bool {
matches!(status, "Completed" | "Failed" | "Aborted")
}
pub fn pid_is_selfware(pid: u32) -> bool {
use sysinfo::{Pid, ProcessesToUpdate, System};
let target = Pid::from_u32(pid);
let mut sys = System::new();
sys.refresh_processes(ProcessesToUpdate::Some(&[target]), true);
match sys.process(target) {
Some(proc_) => {
proc_.name().to_string_lossy().contains("selfware")
|| proc_
.exe()
.map(|e| e.to_string_lossy().contains("selfware"))
.unwrap_or(false)
}
None => false,
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AbortOutcome {
NotFound,
AlreadyDone,
WasStale,
Signalled(u32),
SignalUnsupported(u32),
SignalFailed(u32, String),
}
pub struct RunRegistry {
dir: PathBuf,
}
impl RunRegistry {
pub fn new() -> Result<Self> {
let home = dirs::home_dir().context("cannot resolve home directory")?;
Self::with_dir(home.join(".selfware").join("runs"))
}
pub fn with_dir(dir: PathBuf) -> Result<Self> {
std::fs::create_dir_all(&dir)
.with_context(|| format!("create run registry dir {}", dir.display()))?;
Ok(Self { dir })
}
fn record_path(&self, id: &str) -> PathBuf {
let safe: String = id
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
c
} else {
'_'
}
})
.collect();
self.dir.join(format!("{safe}.json"))
}
pub fn write(&self, record: &RunRecord) -> Result<()> {
let path = self.record_path(&record.id);
let tmp = path.with_extension("json.tmp");
let json = serde_json::to_string_pretty(record).context("serialize run record")?;
std::fs::write(&tmp, json).with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &path).with_context(|| format!("rename into {}", path.display()))?;
Ok(())
}
pub fn get(&self, id: &str) -> Result<Option<RunRecord>> {
let path = self.record_path(id);
if !path.exists() {
return Ok(None);
}
let data =
std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
let rec =
serde_json::from_str(&data).with_context(|| format!("parse {}", path.display()))?;
Ok(Some(rec))
}
pub fn update_status(&self, id: &str, status: &str, now_unix: i64) -> Result<()> {
if let Some(mut rec) = self.get(id)? {
if is_terminal_status(&rec.status) {
return Ok(());
}
rec.status = status.to_string();
rec.updated_unix = now_unix;
self.write(&rec)?;
}
Ok(())
}
pub fn list(&self) -> Result<Vec<RunRecord>> {
let mut out = Vec::new();
for entry in std::fs::read_dir(&self.dir)
.with_context(|| format!("read dir {}", self.dir.display()))?
{
let path = entry?.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
if let Ok(data) = std::fs::read_to_string(&path) {
if let Ok(rec) = serde_json::from_str::<RunRecord>(&data) {
out.push(rec);
}
}
}
out.sort_by_key(|b| std::cmp::Reverse(b.started_unix));
Ok(out)
}
pub fn remove(&self, id: &str) -> Result<()> {
let path = self.record_path(id);
if path.exists() {
std::fs::remove_file(&path).with_context(|| format!("remove {}", path.display()))?;
}
Ok(())
}
pub fn abort(&self, id: &str, now_unix: i64) -> Result<AbortOutcome> {
let Some(mut rec) = self.get(id)? else {
return Ok(AbortOutcome::NotFound);
};
if matches!(rec.status.as_str(), "Completed" | "Failed" | "Aborted") {
return Ok(AbortOutcome::AlreadyDone);
}
let outcome = if pid_alive(rec.pid) && pid_is_selfware(rec.pid) {
#[cfg(unix)]
{
use nix::errno::Errno;
use nix::sys::signal::{kill, Signal};
use nix::unistd::Pid;
match kill(Pid::from_raw(rec.pid as i32), Signal::SIGTERM) {
Ok(()) => AbortOutcome::Signalled(rec.pid),
Err(Errno::ESRCH) => AbortOutcome::WasStale,
Err(e) => return Ok(AbortOutcome::SignalFailed(rec.pid, e.to_string())),
}
}
#[cfg(not(unix))]
{
return Ok(AbortOutcome::SignalUnsupported(rec.pid));
}
} else {
AbortOutcome::WasStale
};
rec.status = "Aborted".to_string();
rec.updated_unix = now_unix;
self.write(&rec)?;
Ok(outcome)
}
}
#[cfg(test)]
#[path = "../../tests/unit/supervision/run_registry/run_registry_test.rs"]
mod tests;