use std::sync::OnceLock;
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub(crate) enum RunStatus {
Running,
Finished,
Failed,
}
impl RunStatus {
fn label(self) -> &'static str {
match self {
RunStatus::Running => "running",
RunStatus::Finished => "finished",
RunStatus::Failed => "failed",
}
}
fn is_terminal(self) -> bool {
matches!(self, RunStatus::Finished | RunStatus::Failed)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum Liveness {
Live,
Ended,
Gone,
Unknown(String),
}
impl Liveness {
pub(crate) fn has_ended(&self) -> bool {
matches!(self, Liveness::Ended | Liveness::Gone)
}
}
const UNDECIDED_GRACE: Duration = Duration::from_secs(5);
#[derive(Default)]
pub(crate) struct UndecidedWatch {
since: Option<Instant>,
reason: String,
}
impl UndecidedWatch {
pub(crate) fn decided(&mut self) {
self.since = None;
self.reason.clear();
}
pub(crate) fn exhausted(&mut self, reason: &str) -> bool {
reason.clone_into(&mut self.reason);
match self.since {
Some(started) if started.elapsed() >= UNDECIDED_GRACE => {
self.since = None;
true
}
Some(_) => false,
None => {
self.since = Some(Instant::now());
false
}
}
}
pub(crate) fn reason(&self) -> &str {
&self.reason
}
}
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub(crate) struct RunDescriptor {
pub(crate) id: String,
pub(crate) pid: u32,
pub(crate) status: RunStatus,
pub(crate) workspace: PathBuf,
pub(crate) plan: PathBuf,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) state_machine: Option<PathBuf>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) control_url: Option<String>,
pub(crate) started_at: String,
pub(crate) headless: bool,
pub(crate) parallel: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) log: Option<PathBuf>,
pub(crate) events: PathBuf,
#[serde(default)]
pub(crate) exit_code: Option<i32>,
}
impl RunDescriptor {
pub(crate) fn liveness(&self) -> Liveness {
let path = run_descriptor_path(&self.workspace);
let current = match read_descriptor_result(&path) {
DescriptorRead::Missing => return Liveness::Gone,
DescriptorRead::Unreadable(why) => {
return Liveness::Unknown(format!("{} could not be read: {why}", path.display()));
}
DescriptorRead::Loaded(current) => current,
};
if current.id != self.id {
return Liveness::Gone;
}
if self.status.is_terminal() || current.status.is_terminal() {
return Liveness::Ended;
}
probe_run_lock(&self.workspace)
}
pub(crate) fn summary_line(&self) -> String {
let plan = self.plan.display();
let mode = if self.headless { "headless" } else { "foreground" };
format!(
"{id} {status:<9} {mode:<10} pid {pid:<7} parallel {parallel} {plan}",
id = self.id,
status = self.status.label(),
pid = self.pid,
parallel = self.parallel,
)
}
}
fn probe_run_lock(workspace_root: &Path) -> Liveness {
let path = workspace_root.join(".rhei").join("run.lock");
let file = match fs::OpenOptions::new().read(true).open(&path) {
Ok(file) => file,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
return Liveness::Unknown(format!("{} does not exist", path.display()));
}
Err(err) => {
return Liveness::Unknown(format!("{} could not be opened: {err}", path.display()));
}
};
match file.try_lock_exclusive() {
Ok(()) => {
let _ = fs2::FileExt::unlock(&file);
Liveness::Ended
}
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => Liveness::Live,
Err(err) => Liveness::Unknown(format!("{} could not be probed: {err}", path.display())),
}
}
pub(crate) fn run_descriptor_path(workspace_root: &Path) -> PathBuf {
workspace_root.join("runtime").join("run.json")
}
pub(crate) fn run_console_log_path(workspace_root: &Path) -> PathBuf {
workspace_root.join("runtime").join("run.log")
}
fn write_descriptor(path: &Path, descriptor: &RunDescriptor) -> std::io::Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let body = serde_json::to_string_pretty(descriptor)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
let temp = path.with_extension("json.tmp");
fs::write(&temp, format!("{body}\n"))?;
fs::rename(&temp, path)
}
pub(crate) enum DescriptorRead {
Loaded(Box<RunDescriptor>),
Missing,
Unreadable(String),
}
pub(crate) fn read_descriptor_result(path: &Path) -> DescriptorRead {
match fs::read_to_string(path) {
Ok(body) => match serde_json::from_str::<RunDescriptor>(&body) {
Ok(descriptor) => DescriptorRead::Loaded(Box::new(descriptor)),
Err(err) => DescriptorRead::Unreadable(err.to_string()),
},
Err(err) if err.kind() == std::io::ErrorKind::NotFound => DescriptorRead::Missing,
Err(err) => DescriptorRead::Unreadable(err.to_string()),
}
}
pub(crate) fn read_descriptor(path: &Path) -> Option<RunDescriptor> {
match read_descriptor_result(path) {
DescriptorRead::Loaded(descriptor) => Some(*descriptor),
DescriptorRead::Missing | DescriptorRead::Unreadable(_) => None,
}
}
struct PublishedRun {
path: PathBuf,
id: String,
pid: u32,
workspace: PathBuf,
}
static PUBLISHED_DESCRIPTOR: OnceLock<Mutex<Option<PublishedRun>>> = OnceLock::new();
fn published_slot() -> &'static Mutex<Option<PublishedRun>> {
PUBLISHED_DESCRIPTOR.get_or_init(|| Mutex::new(None))
}
fn absolutize(path: &Path) -> PathBuf {
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir().map(|cwd| cwd.join(path)).unwrap_or_else(|_| path.to_path_buf())
};
absolute.canonicalize().unwrap_or(absolute)
}
pub(crate) fn publish_run_descriptor(descriptor: &RunDescriptor) {
let descriptor = RunDescriptor {
workspace: absolutize(&descriptor.workspace),
plan: absolutize(&descriptor.plan),
state_machine: descriptor.state_machine.as_deref().map(absolutize),
log: descriptor.log.as_deref().map(absolutize),
events: absolutize(&descriptor.events),
..descriptor.clone()
};
let path = run_descriptor_path(&descriptor.workspace);
if let Err(err) = write_descriptor(&path, &descriptor) {
eprintln!("warning: could not publish the run descriptor at {}: {err}", path.display());
return;
}
*published_slot().lock().unwrap_or_else(|poison| poison.into_inner()) = Some(PublishedRun {
path,
id: descriptor.id.clone(),
pid: descriptor.pid,
workspace: descriptor.workspace.clone(),
});
let Some(registry) = run_registry_path(&descriptor.id) else {
eprintln!(
"warning: neither XDG_STATE_HOME nor HOME is set, so run {} has no registry \
entry; reach it by path instead of by id",
descriptor.id
);
return;
};
if let Err(err) = write_descriptor(®istry, &descriptor) {
eprintln!(
"warning: could not write the run registry entry at {}: {err}\n\
`rhei attach {}` will need the workspace path instead of the id.",
registry.display(),
descriptor.id
);
}
}
pub(crate) fn finalize_run_descriptor(exit_code: i32) {
let taken = published_slot().lock().unwrap_or_else(|poison| poison.into_inner()).take();
let Some(published) = taken else {
return;
};
let Some(mut descriptor) = read_descriptor(&published.path) else {
return;
};
if descriptor.id != published.id || descriptor.pid != published.pid {
return;
}
descriptor.status = if exit_code == 0 { RunStatus::Finished } else { RunStatus::Failed };
descriptor.exit_code = Some(exit_code);
descriptor.control_url = None;
let _ = write_descriptor(&published.path, &descriptor);
finalize_registry_entry(&published, &descriptor);
}
fn finalize_registry_entry(published: &PublishedRun, descriptor: &RunDescriptor) {
let Some(path) = run_registry_path(&published.id) else {
return;
};
let Some(existing) = read_descriptor(&path) else {
return;
};
if existing.pid != published.pid || existing.workspace != published.workspace {
return;
}
let _ = write_descriptor(&path, descriptor);
}