use std::{
path::{Path, PathBuf},
time::Duration,
};
use serde::{Deserialize, Serialize};
const RUNNING_DIR: &str = "running";
const RUNNING_DIR_MODE: u32 = 0o700;
const RECORD_FILE_MODE: u32 = 0o600;
const CONTROL_TOKEN_EXTENSION: &str = "control";
const PROBE_CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
const PROBE_READ_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunningRecord {
pub session_id: String,
pub url: String,
pub pid: u32,
pub process_start: String,
pub capsule_name: String,
pub capsule_version: String,
pub workdir: PathBuf,
pub outlives_launcher: bool,
pub started_at: String,
#[serde(
default,
skip_serializing_if = "Option::is_none",
with = "door_token_field"
)]
pub door_token: Option<crate::door_auth::DoorToken>,
}
mod door_token_field {
use crate::door_auth::DoorToken;
use serde::{Deserialize, Deserializer, Serializer};
pub(super) fn serialize<S: Serializer>(
token: &Option<DoorToken>,
serializer: S,
) -> Result<S::Ok, S::Error> {
match token {
Some(token) => serializer.serialize_some(token.expose()),
None => serializer.serialize_none(),
}
}
pub(super) fn deserialize<'de, D: Deserializer<'de>>(
deserializer: D,
) -> Result<Option<DoorToken>, D::Error> {
Ok(Option::<String>::deserialize(deserializer)?.map(DoorToken::new))
}
}
pub fn running_dir() -> Result<PathBuf, String> {
let dir = crate::murmur_home::ensure_murmur_home()?.join(RUNNING_DIR);
crate::state_store::ensure_private_dir(&dir, RUNNING_DIR_MODE)
.map_err(|reason| format!("{}: {reason}", dir.display()))?;
Ok(dir)
}
pub fn list() -> Result<Vec<RunningRecord>, String> {
let dir = running_dir()?;
let entries = std::fs::read_dir(&dir).map_err(|err| format!("{}: {err}", dir.display()))?;
let mut records = Vec::new();
for entry in entries.flatten() {
let path = entry.path();
if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
continue;
}
match read_record(&path) {
Some(record) => records.push(record),
None => {
let _ = std::fs::remove_file(&path);
}
}
}
records.sort_by(|left, right| right.session_id.cmp(&left.session_id));
Ok(records)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProcessState {
Alive,
Unverified(String),
Gone(String),
}
pub fn process_state(record: &RunningRecord) -> ProcessState {
let alive = pid_is_alive(record.pid);
let token = alive.then(|| process_start_token(record.pid)).flatten();
classify(record, alive, token.as_deref())
}
fn classify(record: &RunningRecord, pid_alive: bool, fresh_token: Option<&str>) -> ProcessState {
if !pid_alive {
return ProcessState::Gone(format!("no process holds pid {}", record.pid));
}
match fresh_token {
Some(token) if token == record.process_start => ProcessState::Alive,
Some(_) => ProcessState::Gone(format!(
"pid {} is held by a process that started at another time",
record.pid
)),
None => {
ProcessState::Unverified(format!("pid {}'s start time could not be read", record.pid))
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Liveness {
Live,
Unreachable(String),
Gone(String),
}
pub fn verify(record: &RunningRecord) -> Liveness {
if let Some(liveness) = liveness_without_probe(process_state(record)) {
return liveness;
}
match probe_session_id(record) {
Ok(session_id) if session_id == record.session_id => Liveness::Live,
Ok(other) => Liveness::Unreachable(format!(
"the capsule at {} answers for session {other}",
record.url
)),
Err(reason) => Liveness::Unreachable(reason),
}
}
fn liveness_without_probe(state: ProcessState) -> Option<Liveness> {
match state {
ProcessState::Alive => None,
ProcessState::Unverified(reason) => Some(Liveness::Unreachable(reason)),
ProcessState::Gone(reason) => Some(Liveness::Gone(reason)),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SignalOutcome {
Sent,
AlreadyGone,
Refused(String),
}
pub fn signal_term(record: &RunningRecord) -> SignalOutcome {
signal(record, libc::SIGTERM)
}
pub fn signal_kill(record: &RunningRecord) -> SignalOutcome {
signal(record, libc::SIGKILL)
}
#[allow(unsafe_code)]
fn signal(record: &RunningRecord, signal: libc::c_int) -> SignalOutcome {
if let Some(outcome) = refusal_before_signal(&process_state(record)) {
return outcome;
}
if unsafe { libc::kill(record.pid as libc::pid_t, signal) } == 0 {
return SignalOutcome::Sent;
}
let err = std::io::Error::last_os_error();
if err.raw_os_error() == Some(libc::ESRCH) {
return SignalOutcome::AlreadyGone;
}
SignalOutcome::Refused(err.to_string())
}
fn refusal_before_signal(state: &ProcessState) -> Option<SignalOutcome> {
match state {
ProcessState::Alive => None,
ProcessState::Unverified(reason) => Some(SignalOutcome::Refused(reason.clone())),
ProcessState::Gone(_) => Some(SignalOutcome::AlreadyGone),
}
}
pub fn prune(record: &RunningRecord) {
if let Ok(dir) = running_dir() {
let _ = std::fs::remove_file(record_path(&dir, &record.session_id));
let _ = std::fs::remove_file(token_path(&dir, &record.session_id));
}
}
pub fn control_token_path(session_id: &str) -> Result<PathBuf, String> {
Ok(token_path(&running_dir()?, session_id))
}
pub struct ControlTokenGuard {
path: PathBuf,
}
impl ControlTokenGuard {
pub fn write(session_id: &str, token: &str) -> Result<Self, String> {
let path = control_token_path(session_id)?;
crate::retention::StagedRewrite::stage_with_mode(
&path,
token.as_bytes(),
Some(RECORD_FILE_MODE),
)
.and_then(crate::retention::StagedRewrite::commit)
.map_err(|reason| format!("failed to write {}: {reason}", path.display()))?;
Ok(Self { path })
}
}
impl Drop for ControlTokenGuard {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.path);
}
}
pub struct RunningGuard {
path: PathBuf,
}
impl RunningGuard {
pub fn write(record: &RunningRecord) -> Result<Self, String> {
let dir = running_dir()?;
let path = record_path(&dir, &record.session_id);
let body = serde_json::to_vec_pretty(record)
.map_err(|err| format!("the record could not be serialized: {err}"))?;
crate::retention::StagedRewrite::stage_with_mode(&path, &body, Some(RECORD_FILE_MODE))
.and_then(crate::retention::StagedRewrite::commit)
.map_err(|reason| format!("failed to write {}: {reason}", path.display()))?;
Ok(Self { path })
}
}
impl Drop for RunningGuard {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.path);
let _ = std::fs::remove_file(self.path.with_extension(CONTROL_TOKEN_EXTENSION));
}
}
#[must_use]
pub fn has_controlling_terminal() -> bool {
std::fs::File::open("/dev/tty").is_ok()
}
fn record_path(dir: &Path, session_id: &str) -> PathBuf {
dir.join(format!("{session_id}.json"))
}
fn token_path(dir: &Path, session_id: &str) -> PathBuf {
dir.join(format!("{session_id}.{CONTROL_TOKEN_EXTENSION}"))
}
fn read_record(path: &Path) -> Option<RunningRecord> {
let body = std::fs::read_to_string(path).ok()?;
serde_json::from_str(&body).ok()
}
#[allow(unsafe_code)]
fn pid_is_alive(pid: u32) -> bool {
let Ok(target) = libc::pid_t::try_from(pid) else {
return false;
};
if target == 0 {
return false;
}
let held = unsafe { libc::kill(target, 0) } == 0
|| std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM);
held && !platform::is_zombie(pid)
}
#[must_use]
pub fn process_start_token(pid: u32) -> Option<String> {
platform::process_start_token(pid)
}
#[cfg(target_os = "linux")]
mod platform {
const STARTTIME_OFFSET_AFTER_COMM: usize = 22 - 3;
pub(super) fn process_start_token(pid: u32) -> Option<String> {
let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
start_token_from_stat(&stat)
}
pub(super) fn start_token_from_stat(stat: &str) -> Option<String> {
fields_after_comm(stat)?
.nth(STARTTIME_OFFSET_AFTER_COMM)
.map(str::to_string)
}
fn fields_after_comm(stat: &str) -> Option<std::str::SplitWhitespace<'_>> {
Some(stat[stat.rfind(')')? + 1..].split_whitespace())
}
pub(super) fn is_zombie(pid: u32) -> bool {
let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else {
return false;
};
fields_after_comm(&stat).and_then(|mut fields| fields.next()) == Some("Z")
}
}
#[cfg(target_os = "macos")]
mod platform {
use std::mem::size_of;
pub(super) fn process_start_token(pid: u32) -> Option<String> {
let info = bsd_info(pid).ok()?;
Some(format!(
"{}.{:06}",
info.pbi_start_tvsec, info.pbi_start_tvusec
))
}
pub(super) fn is_zombie(pid: u32) -> bool {
match bsd_info(pid) {
Ok(info) => info.pbi_status == libc::SZOMB,
Err(err) => err.raw_os_error() == Some(libc::ESRCH),
}
}
#[allow(unsafe_code)]
fn bsd_info(pid: u32) -> std::io::Result<libc::proc_bsdinfo> {
let mut info: libc::proc_bsdinfo = unsafe { std::mem::zeroed() };
let size = size_of::<libc::proc_bsdinfo>();
let written = unsafe {
libc::proc_pidinfo(
pid as libc::c_int,
libc::PROC_PIDTBSDINFO,
0,
std::ptr::addr_of_mut!(info).cast(),
size as libc::c_int,
)
};
if written <= 0 {
return Err(std::io::Error::last_os_error());
}
if usize::try_from(written).ok() != Some(size) {
return Err(std::io::Error::other(format!(
"proc_pidinfo wrote {written} bytes, not one proc_bsdinfo of {size}"
)));
}
Ok(info)
}
}
fn probe_session_id(record: &RunningRecord) -> Result<String, String> {
let addr = record
.url
.trim_start_matches("http://")
.trim_start_matches("https://");
let card = match &record.door_token {
Some(token) => {
let body = serde_json::json!({
"jsonrpc": "2.0",
"id": 1,
"method": crate::identity::DoorMethod::GetAuthenticatedExtendedCard.wire_name(),
"params": {},
})
.to_string();
let authorization = crate::door_auth::bearer_header(token);
let response = crate::http_client::http_json_with_timeouts(
"POST",
&format!("http://{addr}/"),
Some(&body),
&[("Authorization", authorization.as_str())],
PROBE_CONNECT_TIMEOUT,
PROBE_READ_TIMEOUT,
)?;
response.get("result").cloned().ok_or_else(|| {
format!("the capsule at {addr} did not return its extended agent card")
})?
}
None => crate::http_client::http_json_with_timeouts(
"GET",
&format!("http://{addr}/.well-known/agent-card.json"),
None,
&[("Accept", "application/json")],
PROBE_CONNECT_TIMEOUT,
PROBE_READ_TIMEOUT,
)?,
};
crate::identity::session_id_from_card(&card)
.map(str::to_string)
.ok_or_else(|| format!("the agent card from {addr} names no session"))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn prune_removes_the_record_and_its_control_token() {
let home = tempfile::tempdir().unwrap();
crate::murmur_home::run_with_home(
"running::tests::inner_prune_removes_the_record_and_its_control_token",
home.path(),
);
}
#[test]
#[ignore = "run by prune_removes_the_record_and_its_control_token"]
fn inner_prune_removes_the_record_and_its_control_token() {
if !crate::murmur_home::in_scratch_home() {
return;
}
let record = RunningRecord {
session_id: "ses_0199prunecontroltoken000000000000".to_string(),
url: "localhost:1".to_string(),
pid: std::process::id(),
process_start: String::new(),
capsule_name: "c".to_string(),
capsule_version: "0.1.0".to_string(),
workdir: PathBuf::from("/tmp"),
outlives_launcher: false,
started_at: "2026-01-01T00:00:00Z".to_string(),
door_token: None,
};
let token = ControlTokenGuard::write(&record.session_id, "ctl1.token.mac").unwrap();
let record_guard = RunningGuard::write(&record).unwrap();
let dir = running_dir().unwrap();
let token_file = control_token_path(&record.session_id).unwrap();
assert_eq!(
token_file,
dir.join(format!("{}.control", record.session_id))
);
assert_eq!(crate::murmur_home::mode_of(&token_file), RECORD_FILE_MODE);
assert_eq!(
std::fs::read_to_string(&token_file).unwrap(),
"ctl1.token.mac"
);
prune(&record);
assert!(!record_path(&dir, &record.session_id).exists());
assert!(!token_file.exists());
std::mem::forget(token);
std::mem::forget(record_guard);
let _token = ControlTokenGuard::write(&record.session_id, "ctl1.token.mac").unwrap();
drop(RunningGuard::write(&record).unwrap());
assert!(!token_file.exists());
}
fn record(pid: u32, process_start: &str) -> RunningRecord {
RunningRecord {
session_id: "ses_0199c4e2f1b7712a9d3e4f5061728394".to_string(),
url: "127.0.0.1:1".to_string(),
pid,
process_start: process_start.to_string(),
capsule_name: "demo".to_string(),
capsule_version: "0.1.0".to_string(),
workdir: PathBuf::from("/tmp/demo"),
outlives_launcher: true,
started_at: "2026-01-01T00:00:00Z".to_string(),
door_token: None,
}
}
#[test]
fn a_record_serializes_exactly_the_nine_fields() {
let value = serde_json::to_value(record(1, "42")).unwrap();
let mut keys: Vec<&str> = value
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
keys.sort_unstable();
assert_eq!(
keys,
[
"capsule_name",
"capsule_version",
"outlives_launcher",
"pid",
"process_start",
"session_id",
"started_at",
"url",
"workdir",
]
);
}
#[test]
fn a_record_with_a_door_token_serializes_it_beside_the_nine_fields() {
let mut with_token = record(1, "42");
with_token.door_token = Some(crate::door_auth::DoorToken::new(
"mdt1.payload.mac".to_string(),
));
let value = serde_json::to_value(&with_token).unwrap();
let object = value.as_object().unwrap();
assert_eq!(object.len(), 10);
assert_eq!(object["door_token"], "mdt1.payload.mac");
let back: RunningRecord = serde_json::from_value(value).unwrap();
assert_eq!(back, with_token);
}
#[test]
fn a_record_debug_prints_no_door_token() {
let mut with_token = record(1, "42");
with_token.door_token = Some(crate::door_auth::DoorToken::new(
"mdt1.secretpayload.secretmac".to_string(),
));
let debug = format!("{with_token:?}");
assert!(!debug.contains("secretpayload"), "{debug}");
assert!(debug.contains("DoorToken(<redacted>)"), "{debug}");
}
#[test]
fn a_record_without_a_door_token_key_parses() {
let value = serde_json::to_value(record(1, "42")).unwrap();
assert!(value.get("door_token").is_none());
let parsed: RunningRecord = serde_json::from_value(value).unwrap();
assert_eq!(parsed.door_token, None);
}
#[test]
fn a_record_missing_a_field_does_not_parse() {
let mut value = serde_json::to_value(record(1, "42")).unwrap();
value.as_object_mut().unwrap().remove("process_start");
assert!(serde_json::from_value::<RunningRecord>(value).is_err());
}
#[test]
fn this_process_reads_as_alive() {
let pid = std::process::id();
let token = process_start_token(pid).expect("this process has a start time");
assert_eq!(process_state(&record(pid, &token)), ProcessState::Alive);
}
#[test]
fn a_start_time_that_disagrees_reads_as_gone() {
let pid = std::process::id();
let state = process_state(&record(pid, "not-this-process"));
match state {
ProcessState::Gone(reason) => assert!(
reason.contains("another time"),
"the reason must say the pid belongs to another process now: {reason}"
),
other => panic!("a mismatched start time must read as gone, not {other:?}"),
}
}
#[test]
fn a_pid_nothing_holds_reads_as_gone() {
let state = process_state(&record(0, "42"));
assert!(matches!(state, ProcessState::Gone(_)));
}
#[test]
fn the_start_token_is_stable_for_one_process() {
let pid = std::process::id();
assert_eq!(process_start_token(pid), process_start_token(pid));
assert!(process_start_token(pid).is_some_and(|token| !token.is_empty()));
}
#[test]
fn a_zombie_does_not_read_as_alive() {
let mut child = std::process::Command::new("sh")
.args(["-c", "exit 0"])
.spawn()
.expect("sh should start");
let pid = child.id();
let token = process_start_token(pid).expect("a live child has a start time");
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
if matches!(process_state(&record(pid, &token)), ProcessState::Gone(_)) {
break;
}
assert!(
std::time::Instant::now() < deadline,
"a zombie kept reading as alive"
);
std::thread::sleep(Duration::from_millis(50));
}
let _ = child.wait();
}
#[test]
fn a_signal_at_a_mismatched_start_time_is_not_sent() {
let record = record(std::process::id(), "not-this-process");
assert_eq!(signal_kill(&record), SignalOutcome::AlreadyGone);
assert_eq!(signal_term(&record), SignalOutcome::AlreadyGone);
}
#[test]
fn a_signal_at_a_pid_nothing_holds_is_already_gone() {
assert_eq!(signal_term(&record(0, "42")), SignalOutcome::AlreadyGone);
}
#[test]
fn a_live_process_behind_a_silent_port_is_unreachable() {
let pid = std::process::id();
let token = process_start_token(pid).unwrap();
let mut record = record(pid, &token);
let port = {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.local_addr().unwrap().port()
};
record.url = format!("127.0.0.1:{port}");
assert!(matches!(verify(&record), Liveness::Unreachable(_)));
}
const PID: u32 = 4242;
#[test]
fn classify_a_dead_pid_is_gone_whatever_the_token() {
for token in [None, Some("123"), Some("")] {
assert_eq!(
classify(&record(PID, "123"), false, token),
ProcessState::Gone("no process holds pid 4242".to_string()),
"token {token:?}"
);
}
}
#[test]
fn classify_a_live_pid_with_the_recorded_token_is_alive() {
assert_eq!(
classify(&record(PID, "123"), true, Some("123")),
ProcessState::Alive
);
}
#[test]
fn classify_a_live_pid_with_another_token_is_gone() {
assert_eq!(
classify(&record(PID, "123"), true, Some("456")),
ProcessState::Gone("pid 4242 is held by a process that started at another time".into())
);
}
#[test]
fn classify_a_recorded_empty_token_against_a_fresh_one_is_gone() {
assert!(matches!(
classify(&record(PID, ""), true, Some("123")),
ProcessState::Gone(reason) if reason.contains("another time")
));
}
#[test]
fn classify_a_live_pid_whose_token_cannot_be_read_is_unverified() {
assert_eq!(
classify(&record(PID, "123"), true, None),
ProcessState::Unverified("pid 4242's start time could not be read".into())
);
}
#[test]
fn nothing_is_refused_before_signalling_a_confirmed_process() {
assert_eq!(refusal_before_signal(&ProcessState::Alive), None);
}
#[test]
fn a_gone_process_is_already_gone_before_any_signal() {
assert_eq!(
refusal_before_signal(&ProcessState::Gone("no process holds pid 1".into())),
Some(SignalOutcome::AlreadyGone)
);
}
#[test]
fn an_unverified_process_is_refused_before_any_signal() {
let reason = "pid 1's start time could not be read".to_string();
assert_eq!(
refusal_before_signal(&ProcessState::Unverified(reason.clone())),
Some(SignalOutcome::Refused(reason))
);
}
#[test]
fn an_unverified_process_is_unreachable_without_a_probe() {
let reason = "pid 1's start time could not be read".to_string();
assert_eq!(
liveness_without_probe(ProcessState::Unverified(reason.clone())),
Some(Liveness::Unreachable(reason))
);
assert_eq!(
liveness_without_probe(ProcessState::Gone("gone".into())),
Some(Liveness::Gone("gone".into()))
);
assert_eq!(liveness_without_probe(ProcessState::Alive), None);
}
#[cfg(target_os = "linux")]
fn stat_line(comm: &str) -> String {
let fields: Vec<String> = (3..=21).map(|field| field.to_string()).collect();
format!("1234 ({comm}) {} 987654 23 24\n", fields.join(" "))
}
#[cfg(target_os = "linux")]
#[test]
fn the_start_token_is_field_22_of_a_plain_stat_line() {
assert_eq!(
platform::start_token_from_stat(&stat_line("sleep")),
Some("987654".to_string())
);
}
#[cfg(target_os = "linux")]
#[test]
fn the_start_token_is_counted_from_the_last_close_paren() {
assert_eq!(
platform::start_token_from_stat(&stat_line("a) (b c")),
Some("987654".to_string())
);
}
#[cfg(target_os = "linux")]
#[test]
fn a_stat_line_with_no_close_paren_has_no_start_token() {
assert_eq!(platform::start_token_from_stat("1234 sleep S 1 2 3"), None);
}
#[cfg(target_os = "linux")]
#[test]
fn a_stat_line_truncated_before_field_22_has_no_start_token() {
let line = stat_line("sleep");
let truncated = &line[..line.find(" 987654").unwrap()];
assert_eq!(platform::start_token_from_stat(truncated), None);
}
}
#[cfg(test)]
mod home_tests {
use super::*;
#[test]
fn running_dir_holds_a_wide_murmur_home_owner_only() {
let home = tempfile::tempdir().unwrap();
crate::murmur_home::wide_dir(&home.path().join(".murmur"), 0o755);
crate::murmur_home::run_with_home(
"running::home_tests::inner_running_dir_under_scratch_home",
home.path(),
);
assert_eq!(
crate::murmur_home::mode_of(&home.path().join(".murmur")),
0o700
);
assert_eq!(
crate::murmur_home::mode_of(&home.path().join(".murmur").join(RUNNING_DIR)),
RUNNING_DIR_MODE
);
}
#[test]
#[ignore = "run by running_dir_holds_a_wide_murmur_home_owner_only"]
fn inner_running_dir_under_scratch_home() {
if !crate::murmur_home::in_scratch_home() {
return;
}
running_dir().unwrap();
}
}