use std::path::PathBuf;
use sysinfo::{ProcessRefreshKind, ProcessStatus, ProcessesToUpdate, System};
use super::pid_file::PidRecord;
use crate::error::ServerError;
#[derive(Clone, Debug)]
pub struct SelfIdentity {
pub pid: u32,
pub started_at_unix_secs: u64,
pub binary_sha256: String,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum IncarnationProbe {
ProcessGone,
Verified {
exe: Option<PathBuf>,
},
DifferentIncarnation {
live_started_at_unix_secs: u64,
exe: Option<PathBuf>,
},
}
pub fn self_identity() -> Result<SelfIdentity, ServerError> {
let pid = std::process::id();
let started_at_unix_secs =
process_start_instant(pid).ok_or_else(|| ServerError::Incarnation {
message: format!(
"the process table has no entry for this process (pid {pid}); \
cannot record a verifiable incarnation"
),
})?;
let exe = std::env::current_exe().map_err(|io_error| ServerError::Incarnation {
message: format!("cannot resolve this process's executable path: {io_error}"),
})?;
let bytes = std::fs::read(&exe).map_err(|io_error| ServerError::Incarnation {
message: format!(
"cannot read this process's executable `{}` for content hashing: {io_error}",
exe.display()
),
})?;
Ok(SelfIdentity {
pid,
started_at_unix_secs,
binary_sha256: sha256_hex(&bytes),
})
}
#[must_use]
pub fn probe(record: &PidRecord) -> IncarnationProbe {
let mut system = System::new();
let pid = sysinfo::Pid::from_u32(record.pid);
system.refresh_processes_specifics(
ProcessesToUpdate::Some(&[pid]),
true,
ProcessRefreshKind::nothing().with_exe(sysinfo::UpdateKind::Always),
);
let Some(process) = system.process(pid) else {
return IncarnationProbe::ProcessGone;
};
if process.status() == ProcessStatus::Zombie {
return IncarnationProbe::ProcessGone;
}
let live_started_at_unix_secs = process.start_time();
let exe = process.exe().map(std::path::Path::to_path_buf);
if live_started_at_unix_secs == record.started_at_unix_secs {
IncarnationProbe::Verified { exe }
} else {
IncarnationProbe::DifferentIncarnation {
live_started_at_unix_secs,
exe,
}
}
}
fn process_start_instant(pid: u32) -> Option<u64> {
let mut system = System::new();
let pid = sysinfo::Pid::from_u32(pid);
system.refresh_processes_specifics(
ProcessesToUpdate::Some(&[pid]),
true,
ProcessRefreshKind::nothing(),
);
system.process(pid).map(sysinfo::Process::start_time)
}
fn sha256_hex(bytes: &[u8]) -> String {
use sha2::Digest as _;
const HEX: &[u8; 16] = b"0123456789abcdef";
let digest = sha2::Sha256::digest(bytes);
let mut hex = String::with_capacity(digest.len() * 2);
for byte in digest {
hex.push(char::from(HEX[usize::from(byte >> 4)]));
hex.push(char::from(HEX[usize::from(byte & 0x0f)]));
}
hex
}
#[cfg(test)]
mod tests {
use super::{IncarnationProbe, probe, process_start_instant, self_identity};
use crate::control::pid_file::PidRecord;
type TestResult = Result<(), Box<dyn std::error::Error>>;
fn record_with(
pid: u32,
started_at_unix_secs: u64,
) -> Result<PidRecord, std::net::AddrParseError> {
Ok(PidRecord {
pid,
started_at_unix_secs,
binary_sha256: "0".repeat(64),
version: "0.0.0-test".to_owned(),
commit: "test".to_owned(),
state: crate::control::IncarnationState::Serving,
http_address: Some("127.0.0.1:0".parse()?),
grpc_address: Some("127.0.0.1:0".parse()?),
intended_http_address: None,
intended_grpc_address: None,
stage: None,
stage_detail: None,
stage_seq: 0,
stage_updated_at_unix_secs: 0,
drain_timeout_seconds: 30,
})
}
#[test]
fn a_record_of_this_process_verifies() -> TestResult {
let me = self_identity()?;
let record = record_with(me.pid, me.started_at_unix_secs)?;
assert!(
matches!(probe(&record), IncarnationProbe::Verified { .. }),
"this process's own record must verify"
);
Ok(())
}
#[test]
fn a_wrong_start_instant_on_a_live_pid_is_a_different_incarnation() -> TestResult {
let me = self_identity()?;
let record = record_with(me.pid, me.started_at_unix_secs.wrapping_add(7))?;
match probe(&record) {
IncarnationProbe::DifferentIncarnation {
live_started_at_unix_secs,
..
} => {
assert_eq!(
live_started_at_unix_secs, me.started_at_unix_secs,
"the refusal must carry the LIVE process's instant"
);
}
other => {
return Err(format!(
"a mismatched start instant must read as a different incarnation, got {other:?}"
)
.into());
}
}
Ok(())
}
#[test]
fn a_vacant_pid_is_process_gone() -> TestResult {
let child = std::process::Command::new("true").spawn();
let mut child = match child {
Ok(child) => child,
Err(io_error) => return Err(format!("cannot spawn probe child: {io_error}").into()),
};
let pid = child.id();
child.wait()?;
let record = record_with(pid, 1)?;
match probe(&record) {
IncarnationProbe::ProcessGone | IncarnationProbe::DifferentIncarnation { .. } => Ok(()),
IncarnationProbe::Verified { .. } => {
Err("a reaped child's record must never verify".into())
}
}
}
#[test]
fn an_exited_but_unreaped_child_is_process_gone() -> TestResult {
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
let mut child = Command::new("sh")
.arg("-c")
.arg("read _")
.stdin(Stdio::piped())
.spawn()?;
let pid = child.id();
let Some(started_at_unix_secs) = process_start_instant(pid) else {
return Err(format!("probe child {pid} left the table before it was read").into());
};
let record = record_with(pid, started_at_unix_secs)?;
assert!(
matches!(probe(&record), IncarnationProbe::Verified { .. }),
"the specimen only means something if the LIVE child verifies first"
);
drop(child.stdin.take());
let deadline = Instant::now() + Duration::from_secs(10);
let verdict = loop {
let verdict = probe(&record);
if !matches!(verdict, IncarnationProbe::Verified { .. }) || Instant::now() >= deadline {
break verdict;
}
std::thread::sleep(Duration::from_millis(20));
};
let reaped = child.wait();
match verdict {
IncarnationProbe::ProcessGone | IncarnationProbe::DifferentIncarnation { .. } => {
reaped?;
Ok(())
}
IncarnationProbe::Verified { .. } => Err(format!(
"an exited, unreaped child still read as a live incarnation after 10s of \
probing; the reap that followed reported {reaped:?}"
)
.into()),
}
}
}