use std::fs;
use std::io::Write as _;
use std::path::{Path, PathBuf};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use super::discovery::{escaped_name_with_digest, validate_distribution_name};
use crate::paths::AppPaths;
use crate::process::{Adoption, ProcessIdentity};
use runner_manager_domain::model::ScaleTarget;
pub const GUEST_CONFIG_FILE: &str = "wsl-recovery.toml";
pub const REQUEST_FILE: &str = "drain-request.json";
pub const HEARTBEAT_FILE: &str = "guest-heartbeat.json";
pub const FENCE_DIRECTORY: &str = "launch-fence";
pub const OWNER_FILE: &str = "owner.json";
pub const RECOVERY_STATUS_FILE: &str = "recovery-status.json";
pub const SCHEMA_VERSION: u32 = 1;
pub fn recovery_root(paths: &AppPaths, distribution: &str) -> Result<PathBuf, super::WslError> {
validate_distribution_name(distribution)?;
Ok(paths
.config_dir()
.join("wsl-recovery")
.join(escaped_name_with_digest(distribution)))
}
#[derive(Debug, thiserror::Error)]
pub enum FenceError {
#[error("cannot {operation} WSL recovery state at {}: {source}", path.display())]
Io {
operation: &'static str,
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("cannot decode WSL recovery state at {}: {source}", path.display())]
Decode {
path: PathBuf,
#[source]
source: serde_json::Error,
},
#[error("WSL recovery state at {} has schema {found}, but this build supports {SCHEMA_VERSION}", path.display())]
Schema { path: PathBuf, found: u32 },
}
fn io(operation: &'static str, path: &Path, source: std::io::Error) -> FenceError {
FenceError::Io {
operation,
path: path.to_path_buf(),
source,
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GuestRecoveryConfig {
pub schema_version: u32,
pub shared_root: PathBuf,
}
impl GuestRecoveryConfig {
#[must_use]
pub fn new(shared_root: PathBuf) -> Self {
Self {
schema_version: SCHEMA_VERSION,
shared_root,
}
}
#[must_use]
pub fn path(paths: &AppPaths) -> PathBuf {
paths.config_dir().join(GUEST_CONFIG_FILE)
}
pub fn read(paths: &AppPaths) -> Result<Option<Self>, FenceError> {
let path = Self::path(paths);
let text = match fs::read_to_string(&path) {
Ok(text) => text,
Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(source) => return Err(io("read", &path, source)),
};
let value: Self = toml::from_str(&text).map_err(|source| FenceError::Io {
operation: "decode",
path: path.clone(),
source: std::io::Error::new(std::io::ErrorKind::InvalidData, source),
})?;
if value.schema_version != SCHEMA_VERSION {
return Err(FenceError::Schema {
path,
found: value.schema_version,
});
}
Ok(Some(value))
}
pub fn write(&self, paths: &AppPaths) -> Result<(), FenceError> {
let path = Self::path(paths);
let text = toml::to_string_pretty(self).map_err(|source| FenceError::Io {
operation: "encode",
path: path.clone(),
source: std::io::Error::new(std::io::ErrorKind::InvalidData, source),
})?;
atomic_write(&path, text.as_bytes())
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DrainRequest {
pub schema_version: u32,
pub generation: u64,
pub requested_at: DateTime<Utc>,
}
impl DrainRequest {
#[must_use]
pub fn new(generation: u64, requested_at: DateTime<Utc>) -> Self {
Self {
schema_version: SCHEMA_VERSION,
generation,
requested_at,
}
}
pub fn write(&self, root: &Path) -> Result<(), FenceError> {
write_json(&root.join(REQUEST_FILE), self)
}
pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
read_json(&root.join(REQUEST_FILE))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GuestHeartbeat {
pub schema_version: u32,
pub observed_at: DateTime<Utc>,
pub acknowledged_generation: Option<u64>,
pub local_active_attempts: Option<u32>,
#[serde(default)]
pub local_busy_attempts: Option<u32>,
pub managed_targets: Vec<ScaleTarget>,
pub unmanaged_runner_services: Option<u32>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RecoveryPhase {
Healthy,
Degraded,
Draining,
Recovering,
Backoff,
RecoveryBlocked,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RecoveryStatus {
pub schema_version: u32,
pub observed_at: DateTime<Utc>,
pub phase: RecoveryPhase,
pub consecutive_probe_failures: u8,
pub reason: Option<String>,
pub last_recovered_at: Option<DateTime<Utc>>,
}
impl RecoveryStatus {
pub fn write(&self, root: &Path) -> Result<(), FenceError> {
write_json(&root.join(RECOVERY_STATUS_FILE), self)
}
pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
read_json(&root.join(RECOVERY_STATUS_FILE))
}
}
impl GuestHeartbeat {
pub fn write(&self, root: &Path) -> Result<(), FenceError> {
write_json(&root.join(HEARTBEAT_FILE), self)
}
pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
read_json(&root.join(HEARTBEAT_FILE))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FenceOwnerKind {
GuestLaunch,
WindowsRecovery,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FenceOwner {
pub schema_version: u32,
pub kind: FenceOwnerKind,
pub generation: Option<u64>,
pub process_id: u32,
pub acquired_at: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub identity: Option<ProcessIdentity>,
}
#[derive(Debug)]
pub struct FenceClaim {
directory: PathBuf,
release_on_drop: bool,
}
impl FenceClaim {
pub fn try_claim(
root: &Path,
kind: FenceOwnerKind,
generation: Option<u64>,
) -> Result<Option<Self>, FenceError> {
fs::create_dir_all(root).map_err(|source| io("create", root, source))?;
let directory = root.join(FENCE_DIRECTORY);
match fs::create_dir(&directory) {
Ok(()) => {}
Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => return Ok(None),
Err(source) => return Err(io("claim", &directory, source)),
}
let owner = FenceOwner {
schema_version: SCHEMA_VERSION,
kind,
generation,
process_id: std::process::id(),
acquired_at: Utc::now(),
identity: ProcessIdentity::of_current_process().ok(),
};
if let Err(error) = write_json(&directory.join(OWNER_FILE), &owner) {
let _ = fs::remove_dir_all(&directory);
return Err(error);
}
Ok(Some(Self {
directory,
release_on_drop: true,
}))
}
pub fn make_durable(mut self) {
self.release_on_drop = false;
}
pub fn owner(root: &Path) -> Result<Option<FenceOwner>, FenceError> {
read_json(&root.join(FENCE_DIRECTORY).join(OWNER_FILE))
}
pub fn release(mut self) -> Result<(), FenceError> {
self.release_on_drop = false;
remove_claim(&self.directory)
}
}
impl Drop for FenceClaim {
fn drop(&mut self) {
if self.release_on_drop {
let _ = remove_claim(&self.directory);
}
}
}
#[must_use]
pub fn elapsed_at_least(
since: DateTime<Utc>,
now: DateTime<Utc>,
limit: std::time::Duration,
) -> bool {
(now - since).to_std().is_ok_and(|age| age >= limit)
}
pub(crate) fn fence_modified(root: &Path) -> Result<Option<DateTime<Utc>>, FenceError> {
let directory = root.join(FENCE_DIRECTORY);
match fs::metadata(&directory).and_then(|meta| meta.modified()) {
Ok(modified) => Ok(Some(modified.into())),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(source) => Err(io("inspect", &directory, source)),
}
}
const RELEASE_ATTEMPTS: u32 = 3;
const RELEASE_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StaleGuestClaim {
OwnerExited,
EarlierBoot,
WrittenBeforeThisDaemon,
ReleaseFailed,
OwnerNeverRecorded,
}
impl std::fmt::Display for StaleGuestClaim {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
Self::OwnerExited => "owner_exited",
Self::EarlierBoot => "earlier_boot",
Self::WrittenBeforeThisDaemon => "written_before_this_daemon",
Self::ReleaseFailed => "release_failed",
Self::OwnerNeverRecorded => "owner_never_recorded",
})
}
}
pub(crate) const OWNERLESS_CLAIM_GRACE: std::time::Duration =
std::time::Duration::from_secs(5 * 60);
pub(crate) trait ProcessProbe {
fn current(&self) -> Option<ProcessIdentity>;
fn recheck(&self, identity: &ProcessIdentity) -> Option<Adoption>;
fn pid_alive(&self, pid: u32) -> Option<bool>;
fn boot_time(&self) -> Option<DateTime<Utc>>;
fn process_started(&self) -> DateTime<Utc>;
}
pub(crate) struct HostProcesses;
impl ProcessProbe for HostProcesses {
fn current(&self) -> Option<ProcessIdentity> {
ProcessIdentity::of_current_process().ok()
}
fn recheck(&self, identity: &ProcessIdentity) -> Option<Adoption> {
identity.recheck().ok()
}
fn pid_alive(&self, pid: u32) -> Option<bool> {
match ProcessIdentity::resolve(pid) {
Ok(_) => Some(true),
Err(crate::process::ProcessError::NoSuchProcess { .. }) => Some(false),
Err(_) => None,
}
}
fn boot_time(&self) -> Option<DateTime<Utc>> {
crate::process::boot_time()
}
fn process_started(&self) -> DateTime<Utc> {
static STARTED: std::sync::LazyLock<DateTime<Utc>> = std::sync::LazyLock::new(Utc::now);
*STARTED
}
}
fn stale_guest_owner(owner: &FenceOwner, probe: &dyn ProcessProbe) -> Option<StaleGuestClaim> {
if owner.kind != FenceOwnerKind::GuestLaunch {
return None;
}
let current = probe.current();
if let Some(identity) = &owner.identity {
if current.as_ref() == Some(identity) {
return Some(StaleGuestClaim::ReleaseFailed);
}
return match probe.recheck(identity)? {
Adoption::Live => None,
Adoption::Gone | Adoption::PidRecycled { .. } => Some(StaleGuestClaim::OwnerExited),
};
}
if probe
.boot_time()
.is_some_and(|booted| owner.acquired_at < booted)
{
return Some(StaleGuestClaim::EarlierBoot);
}
if owner.acquired_at < probe.process_started() {
return Some(StaleGuestClaim::WrittenBeforeThisDaemon);
}
if current.is_some_and(|me| me.pid() == owner.process_id) {
return Some(StaleGuestClaim::ReleaseFailed);
}
match probe.pid_alive(owner.process_id)? {
true => None,
false => Some(StaleGuestClaim::OwnerExited),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReclaimedGuestClaim {
pub reason: StaleGuestClaim,
pub owner: Option<FenceOwner>,
}
impl std::fmt::Display for ReclaimedGuestClaim {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.reason)?;
if let Some(owner) = &self.owner {
write!(
f,
" (pid {}, acquired {})",
owner.process_id,
owner.acquired_at.to_rfc3339()
)?;
}
Ok(())
}
}
static HELD_HERE: std::sync::Mutex<std::collections::BTreeSet<PathBuf>> =
std::sync::Mutex::new(std::collections::BTreeSet::new());
fn held_key(root: &Path) -> PathBuf {
fs::canonicalize(root)
.unwrap_or_else(|_| root.to_path_buf())
.join(FENCE_DIRECTORY)
}
fn held_here() -> std::sync::MutexGuard<'static, std::collections::BTreeSet<PathBuf>> {
HELD_HERE
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
#[derive(Debug)]
pub enum GuestClaimAttempt {
Claimed {
claim: GuestLaunchClaim,
reclaimed: Option<ReclaimedGuestClaim>,
},
Busy,
}
#[derive(Debug)]
pub struct GuestLaunchClaim {
claim: Option<FenceClaim>,
key: PathBuf,
}
impl GuestLaunchClaim {
pub fn release(mut self) -> Result<(), FenceError> {
self.release_now()
}
fn release_now(&mut self) -> Result<(), FenceError> {
let Some(mut claim) = self.claim.take() else {
return Ok(());
};
claim.release_on_drop = false;
let mut held = held_here();
held.remove(&self.key);
let mut result = Ok(());
for attempt in 1..=RELEASE_ATTEMPTS {
result = remove_claim(&claim.directory);
if result.is_ok() || attempt == RELEASE_ATTEMPTS {
break;
}
std::thread::sleep(RELEASE_RETRY_DELAY * attempt);
}
result
}
}
impl Drop for GuestLaunchClaim {
fn drop(&mut self) {
let _ = self.release_now();
}
}
pub fn try_claim_guest_launch(root: &Path) -> Result<GuestClaimAttempt, FenceError> {
try_claim_guest_launch_with(root, &HostProcesses, Utc::now())
}
pub(crate) fn try_claim_guest_launch_with(
root: &Path,
probe: &dyn ProcessProbe,
now: DateTime<Utc>,
) -> Result<GuestClaimAttempt, FenceError> {
fs::create_dir_all(root).map_err(|source| io("create", root, source))?;
let key = held_key(root);
let mut held = held_here();
if held.contains(&key) {
return Ok(GuestClaimAttempt::Busy);
}
let mut claim = FenceClaim::try_claim(root, FenceOwnerKind::GuestLaunch, None)?;
let mut reclaimed = None;
if claim.is_none() {
reclaimed = reclaim_unheld(root, probe, now)?;
if reclaimed.is_some() {
claim = FenceClaim::try_claim(root, FenceOwnerKind::GuestLaunch, None)?;
}
}
let Some(claim) = claim else {
return Ok(GuestClaimAttempt::Busy);
};
held.insert(key.clone());
Ok(GuestClaimAttempt::Claimed {
claim: GuestLaunchClaim {
claim: Some(claim),
key,
},
reclaimed,
})
}
pub fn reclaim_stale_guest_claim(root: &Path) -> Result<Option<ReclaimedGuestClaim>, FenceError> {
reclaim_stale_guest_claim_with(root, &HostProcesses, Utc::now())
}
pub(crate) fn reclaim_stale_guest_claim_with(
root: &Path,
probe: &dyn ProcessProbe,
now: DateTime<Utc>,
) -> Result<Option<ReclaimedGuestClaim>, FenceError> {
let held = held_here();
if held.contains(&held_key(root)) {
return Ok(None);
}
reclaim_unheld(root, probe, now)
}
fn reclaim_unheld(
root: &Path,
probe: &dyn ProcessProbe,
now: DateTime<Utc>,
) -> Result<Option<ReclaimedGuestClaim>, FenceError> {
let directory = root.join(FENCE_DIRECTORY);
let Some(owner) = FenceClaim::owner(root)? else {
let abandoned = fence_modified(root)?
.is_some_and(|modified| elapsed_at_least(modified, now, OWNERLESS_CLAIM_GRACE));
if !abandoned || FenceClaim::owner(root)?.is_some() {
return Ok(None);
}
remove_claim(&directory)?;
return Ok(Some(ReclaimedGuestClaim {
reason: StaleGuestClaim::OwnerNeverRecorded,
owner: None,
}));
};
let Some(reason) = stale_guest_owner(&owner, probe) else {
return Ok(None);
};
if FenceClaim::owner(root)?.as_ref() != Some(&owner) {
return Ok(None);
}
remove_claim(&directory)?;
Ok(Some(ReclaimedGuestClaim {
reason,
owner: Some(owner),
}))
}
pub fn clear_recovery(root: &Path, generation: u64) -> Result<(), FenceError> {
let owner = FenceClaim::owner(root)?;
if owner.as_ref().is_some_and(|owner| {
owner.kind == FenceOwnerKind::WindowsRecovery && owner.generation == Some(generation)
}) {
remove_claim(&root.join(FENCE_DIRECTORY))?;
}
let request_path = root.join(REQUEST_FILE);
if DrainRequest::read(root)?
.as_ref()
.is_some_and(|request| request.generation != generation)
{
return Ok(());
}
match fs::remove_file(&request_path) {
Ok(()) => Ok(()),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(source) => Err(io("remove", &request_path, source)),
}
}
pub fn retire_windows_recovery(root: &Path) -> Result<(), FenceError> {
let recovery_generation = FenceClaim::owner(root)?.and_then(|owner| {
(owner.kind == FenceOwnerKind::WindowsRecovery)
.then_some(owner.generation)
.flatten()
});
if let Some(generation) = recovery_generation {
clear_recovery(root, generation)?;
}
if let Some(request) = DrainRequest::read(root)? {
clear_recovery(root, request.generation)?;
}
let status_path = root.join(RECOVERY_STATUS_FILE);
match fs::remove_file(&status_path) {
Ok(()) => Ok(()),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(source) => Err(io("remove", &status_path, source)),
}
}
pub fn retire_abandoned_windows_recovery(
root: &Path,
now: DateTime<Utc>,
abandoned_after: std::time::Duration,
) -> Result<bool, FenceError> {
let abandoned = |at: DateTime<Utc>| elapsed_at_least(at, now, abandoned_after);
let request = DrainRequest::read(root)?;
if request
.as_ref()
.is_some_and(|request| !abandoned(request.requested_at))
{
return Ok(false);
}
let windows_owner =
FenceClaim::owner(root)?.filter(|owner| owner.kind == FenceOwnerKind::WindowsRecovery);
if windows_owner
.as_ref()
.is_some_and(|owner| !abandoned(owner.acquired_at))
{
return Ok(false);
}
if request.is_none() && windows_owner.is_none() {
return Ok(false);
}
retire_windows_recovery(root)?;
Ok(true)
}
fn remove_claim(directory: &Path) -> Result<(), FenceError> {
match fs::remove_dir_all(directory) {
Ok(()) => Ok(()),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(source) => Err(io("release", directory, source)),
}
}
fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), FenceError> {
let parent = path.parent().unwrap_or_else(|| Path::new("."));
fs::create_dir_all(parent).map_err(|source| io("create", parent, source))?;
let mut temporary =
tempfile::NamedTempFile::new_in(parent).map_err(|source| io("write", path, source))?;
temporary
.write_all(bytes)
.and_then(|()| temporary.as_file().sync_all())
.map_err(|source| io("write", path, source))?;
temporary
.persist(path)
.map(|_| ())
.map_err(|error| io("replace", path, error.error))
}
fn write_json<T: Serialize>(path: &Path, value: &T) -> Result<(), FenceError> {
let bytes = serde_json::to_vec_pretty(value).map_err(|source| FenceError::Decode {
path: path.to_path_buf(),
source,
})?;
atomic_write(path, &bytes)
}
fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<Option<T>, FenceError> {
let bytes = match fs::read(path) {
Ok(bytes) => bytes,
Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(source) => return Err(io("read", path, source)),
};
let value: T = serde_json::from_slice(&bytes).map_err(|source| FenceError::Decode {
path: path.to_path_buf(),
source,
})?;
Ok(Some(value))
}
#[must_use]
pub fn unmanaged_runner_service_count() -> Option<u32> {
if !cfg!(target_os = "linux") {
return Some(0);
}
let mut names = std::collections::BTreeSet::new();
for directory in [
"/etc/systemd/system",
"/usr/lib/systemd/system",
"/lib/systemd/system",
] {
let entries = match fs::read_dir(directory) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
Err(_) => return None,
};
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().into_owned();
if name.starts_with("actions.runner.") && name.ends_with(".service") {
names.insert(name);
}
}
}
u32::try_from(names.len()).ok()
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Default)]
struct FakeProbe {
current: Option<ProcessIdentity>,
recheck: Option<Adoption>,
pid_alive: Option<bool>,
boot_time: Option<DateTime<Utc>>,
process_started: Option<DateTime<Utc>>,
}
impl ProcessProbe for FakeProbe {
fn current(&self) -> Option<ProcessIdentity> {
self.current.clone()
}
fn recheck(&self, _: &ProcessIdentity) -> Option<Adoption> {
self.recheck.clone()
}
fn pid_alive(&self, _: u32) -> Option<bool> {
self.pid_alive
}
fn boot_time(&self) -> Option<DateTime<Utc>> {
self.boot_time
}
fn process_started(&self) -> DateTime<Utc> {
self.process_started.unwrap_or(DateTime::<Utc>::MIN_UTC)
}
}
fn identity(pid: u32, token: &str) -> ProcessIdentity {
serde_json::from_value(serde_json::json!({ "pid": pid, "start_token": token })).unwrap()
}
fn guest_owner(
process_id: u32,
acquired_at: DateTime<Utc>,
identity: Option<ProcessIdentity>,
) -> FenceOwner {
FenceOwner {
schema_version: SCHEMA_VERSION,
kind: FenceOwnerKind::GuestLaunch,
generation: None,
process_id,
acquired_at,
identity,
}
}
fn plant(root: &Path, owner: &FenceOwner) {
fs::create_dir_all(root.join(FENCE_DIRECTORY)).unwrap();
write_json(&root.join(FENCE_DIRECTORY).join(OWNER_FILE), owner).unwrap();
}
fn claimed(attempt: GuestClaimAttempt) -> (GuestLaunchClaim, Option<ReclaimedGuestClaim>) {
match attempt {
GuestClaimAttempt::Claimed { claim, reclaimed } => (claim, reclaimed),
GuestClaimAttempt::Busy => panic!("the launch was refused"),
}
}
fn exited_child() -> u32 {
let mut child = crate::process::tests::quick_exit().spawn().unwrap();
child.wait().unwrap();
child.pid()
}
#[test]
fn a_guest_claim_whose_owner_exited_is_reclaimed_and_the_launch_proceeds() {
let root = tempfile::tempdir().unwrap();
let pid = exited_child();
plant(
root.path(),
&guest_owner(pid, Utc::now(), Some(identity(pid, "platform:gone"))),
);
let (claim, reclaimed) =
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
let reclaimed = reclaimed.expect("the dead owner's claim was reclaimed");
assert_eq!(reclaimed.reason, StaleGuestClaim::OwnerExited);
assert_eq!(reclaimed.owner.unwrap().process_id, pid);
let owner = FenceClaim::owner(root.path()).unwrap().unwrap();
assert_eq!(
owner.identity,
HostProcesses.current(),
"the new claim is ours"
);
drop(claim);
assert!(!root.path().join(FENCE_DIRECTORY).exists());
}
#[test]
fn a_guest_claim_whose_pid_now_belongs_to_another_process_is_reclaimed() {
let root = tempfile::tempdir().unwrap();
let pid = std::process::id();
plant(
root.path(),
&guest_owner(
pid,
Utc::now(),
Some(identity(pid, "linux:earlier-boot:189")),
),
);
let (_claim, reclaimed) =
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
assert_eq!(reclaimed.unwrap().reason, StaleGuestClaim::OwnerExited);
}
#[test]
fn a_live_guest_claim_is_never_reclaimed() {
let root = tempfile::tempdir().unwrap();
let mut child = crate::process::tests::long_running().spawn().unwrap();
let live = child.identity().clone();
assert_eq!(
HostProcesses.recheck(&live),
Some(Adoption::Live),
"the owner must be known live, not merely uninspectable"
);
plant(
root.path(),
&guest_owner(child.pid(), Utc::now(), Some(live.clone())),
);
let attempt = try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now());
let startup = reclaim_stale_guest_claim_with(root.path(), &HostProcesses, Utc::now());
live.terminate(std::time::Duration::from_secs(5)).unwrap();
child.wait().unwrap();
assert!(matches!(attempt.unwrap(), GuestClaimAttempt::Busy));
assert_eq!(startup.unwrap(), None);
assert_eq!(
FenceClaim::owner(root.path()).unwrap().unwrap().identity,
Some(live)
);
}
#[test]
fn a_claim_this_process_holds_refuses_every_other_launch_of_it() {
let root = tempfile::tempdir().unwrap();
let (claim, reclaimed) =
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
assert_eq!(reclaimed, None);
assert!(matches!(
try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap(),
GuestClaimAttempt::Busy
));
assert_eq!(
reclaim_stale_guest_claim_with(root.path(), &HostProcesses, Utc::now()).unwrap(),
None
);
drop(claim);
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
}
#[test]
fn a_claim_this_process_failed_to_release_is_reclaimed_by_it() {
let root = tempfile::tempdir().unwrap();
let (mut claim, _) =
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
claim.claim.take().unwrap().make_durable();
held_here().remove(&held_key(root.path()));
drop(claim);
assert!(root.path().join(FENCE_DIRECTORY).exists());
let (_claim, reclaimed) =
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
assert_eq!(reclaimed.unwrap().reason, StaleGuestClaim::ReleaseFailed);
}
#[test]
fn a_windows_recovery_claim_is_never_reclaimed_by_the_guest() {
let root = tempfile::tempdir().unwrap();
let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(3))
.unwrap()
.unwrap();
claim.make_durable();
let dead = FakeProbe {
recheck: Some(Adoption::Gone),
pid_alive: Some(false),
boot_time: Some(Utc::now() + chrono::Duration::days(1)),
..FakeProbe::default()
};
assert!(matches!(
try_claim_guest_launch_with(root.path(), &dead, Utc::now()).unwrap(),
GuestClaimAttempt::Busy
));
assert!(root.path().join(FENCE_DIRECTORY).exists());
}
const PRE_0_4_35_RECORD: &str = r#"{
"schema_version": 1,
"kind": "guest_launch",
"generation": null,
"process_id": 189,
"acquired_at": "2026-09-24T00:47:18.724Z"
}"#;
#[test]
fn an_old_schema_record_is_judged_by_the_boot_and_the_pid() {
let old: FenceOwner = serde_json::from_str(PRE_0_4_35_RECORD).unwrap();
assert_eq!(old.identity, None);
let after = old.acquired_at + chrono::Duration::hours(1);
let before = old.acquired_at - chrono::Duration::hours(1);
let me = Some(identity(4242, "platform:me"));
let rebooted = FakeProbe {
boot_time: Some(after),
pid_alive: Some(true),
..FakeProbe::default()
};
assert_eq!(
stale_guest_owner(&old, &rebooted),
Some(StaleGuestClaim::EarlierBoot)
);
let exited = FakeProbe {
boot_time: Some(before),
pid_alive: Some(false),
..FakeProbe::default()
};
assert_eq!(
stale_guest_owner(&old, &exited),
Some(StaleGuestClaim::OwnerExited)
);
let alive = FakeProbe {
boot_time: Some(before),
pid_alive: Some(true),
current: me.clone(),
..FakeProbe::default()
};
assert_eq!(stale_guest_owner(&old, &alive), None);
let unknown = FakeProbe::default();
assert_eq!(stale_guest_owner(&old, &unknown), None);
let distribution_restarted = FakeProbe {
boot_time: Some(before),
pid_alive: Some(true),
process_started: Some(after),
..FakeProbe::default()
};
assert_eq!(
stale_guest_owner(&old, &distribution_restarted),
Some(StaleGuestClaim::WrittenBeforeThisDaemon)
);
let mine = FakeProbe {
boot_time: Some(before),
pid_alive: Some(true),
current: Some(identity(189, "platform:me")),
..FakeProbe::default()
};
assert_eq!(
stale_guest_owner(&old, &mine),
Some(StaleGuestClaim::ReleaseFailed)
);
}
#[test]
fn the_incident_record_is_reclaimed_at_daemon_start_after_an_update() {
let root = tempfile::tempdir().unwrap();
fs::create_dir_all(root.path().join(FENCE_DIRECTORY)).unwrap();
fs::write(
root.path().join(FENCE_DIRECTORY).join(OWNER_FILE),
PRE_0_4_35_RECORD,
)
.unwrap();
let same_boot_new_daemon = FakeProbe {
boot_time: Some("2026-09-20T00:00:00Z".parse().unwrap()),
pid_alive: Some(true),
process_started: Some("2026-10-07T18:00:00Z".parse().unwrap()),
..FakeProbe::default()
};
let reclaimed =
reclaim_stale_guest_claim_with(root.path(), &same_boot_new_daemon, Utc::now())
.unwrap()
.unwrap();
assert_eq!(reclaimed.reason, StaleGuestClaim::WrittenBeforeThisDaemon);
assert!(!root.path().join(FENCE_DIRECTORY).exists());
claimed(
try_claim_guest_launch_with(root.path(), &same_boot_new_daemon, Utc::now()).unwrap(),
);
}
#[cfg(target_os = "linux")]
#[test]
fn this_linux_host_reports_when_it_booted() {
let booted = HostProcesses
.boot_time()
.expect("/proc/stat names the boot");
assert!(booted < Utc::now(), "{booted}");
}
#[test]
fn an_ownerless_claim_is_reclaimed_only_after_the_grace() {
let root = tempfile::tempdir().unwrap();
fs::create_dir_all(root.path().join(FENCE_DIRECTORY)).unwrap();
let probe = FakeProbe::default();
assert_eq!(
reclaim_stale_guest_claim_with(root.path(), &probe, Utc::now()).unwrap(),
None
);
let later = Utc::now() + chrono::Duration::minutes(6);
let reclaimed = reclaim_stale_guest_claim_with(root.path(), &probe, later)
.unwrap()
.unwrap();
assert_eq!(reclaimed.reason, StaleGuestClaim::OwnerNeverRecorded);
assert!(!root.path().join(FENCE_DIRECTORY).exists());
}
#[test]
fn a_release_that_fails_reports_its_path_and_the_next_launch_reclaims_it() {
let root = tempfile::tempdir().unwrap();
let (claim, _) =
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
let directory = root.path().join(FENCE_DIRECTORY);
let owner = fs::read(directory.join(OWNER_FILE)).unwrap();
fs::remove_dir_all(&directory).unwrap();
fs::write(&directory, b"not a directory").unwrap();
assert!(
remove_claim(&directory).is_err(),
"the plant must make removal fail"
);
let error = claim.release().expect_err("the removal failed");
assert!(error.to_string().contains(FENCE_DIRECTORY), "{error}");
fs::remove_file(&directory).unwrap();
fs::create_dir(&directory).unwrap();
fs::write(directory.join(OWNER_FILE), owner).unwrap();
let (_claim, reclaimed) =
claimed(try_claim_guest_launch_with(root.path(), &HostProcesses, Utc::now()).unwrap());
assert_eq!(reclaimed.unwrap().reason, StaleGuestClaim::ReleaseFailed);
}
#[test]
fn one_directory_has_exactly_one_owner_and_drop_releases_guest_claim() {
let root = tempfile::tempdir().unwrap();
let first = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
.unwrap()
.unwrap();
assert!(
FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
.unwrap()
.is_none()
);
drop(first);
assert!(
FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
.unwrap()
.is_some()
);
}
#[test]
fn only_matching_recovery_generation_is_cleared() {
let root = tempfile::tempdir().unwrap();
let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(9))
.unwrap()
.unwrap();
claim.make_durable();
DrainRequest::new(9, Utc::now()).write(root.path()).unwrap();
clear_recovery(root.path(), 8).unwrap();
assert!(root.path().join(FENCE_DIRECTORY).exists());
assert_eq!(
DrainRequest::read(root.path()).unwrap().unwrap().generation,
9
);
clear_recovery(root.path(), 9).unwrap();
assert!(!root.path().join(FENCE_DIRECTORY).exists());
assert!(!root.path().join(REQUEST_FILE).exists());
}
#[test]
fn retiring_windows_recovery_handles_restarted_generations_but_keeps_guest_claims() {
let root = tempfile::tempdir().unwrap();
let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(7))
.unwrap()
.unwrap();
claim.make_durable();
DrainRequest::new(8, Utc::now()).write(root.path()).unwrap();
RecoveryStatus {
schema_version: SCHEMA_VERSION,
observed_at: Utc::now(),
phase: RecoveryPhase::RecoveryBlocked,
consecutive_probe_failures: 9,
reason: Some("stale".into()),
last_recovered_at: None,
}
.write(root.path())
.unwrap();
retire_windows_recovery(root.path()).unwrap();
assert!(!root.path().join(FENCE_DIRECTORY).exists());
assert!(!root.path().join(REQUEST_FILE).exists());
assert!(!root.path().join(RECOVERY_STATUS_FILE).exists());
let guest = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
.unwrap()
.unwrap();
guest.make_durable();
DrainRequest::new(10, Utc::now())
.write(root.path())
.unwrap();
retire_windows_recovery(root.path()).unwrap();
assert!(root.path().join(FENCE_DIRECTORY).exists());
assert!(!root.path().join(REQUEST_FILE).exists());
}
#[test]
fn a_restarted_watchdog_retires_only_recovery_nobody_is_still_driving() {
let root = tempfile::tempdir().unwrap();
let now = Utc::now();
let window = std::time::Duration::from_secs(300);
let old = now - chrono::Duration::minutes(10);
let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
.unwrap()
.unwrap();
claim.make_durable();
let owner_path = root.path().join(FENCE_DIRECTORY).join(OWNER_FILE);
let backdate_owner = || {
let mut owner = FenceClaim::owner(root.path()).unwrap().unwrap();
owner.acquired_at = old;
write_json(&owner_path, &owner).unwrap();
};
backdate_owner();
DrainRequest::new(2, now - chrono::Duration::seconds(10))
.write(root.path())
.unwrap();
assert!(!retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
assert!(root.path().join(FENCE_DIRECTORY).exists());
DrainRequest::new(2, old).write(root.path()).unwrap();
assert!(retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
assert!(!root.path().join(FENCE_DIRECTORY).exists());
assert!(!root.path().join(REQUEST_FILE).exists());
assert!(!retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(3))
.unwrap()
.unwrap();
claim.make_durable();
assert!(!retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
assert!(root.path().join(FENCE_DIRECTORY).exists());
remove_claim(&root.path().join(FENCE_DIRECTORY)).unwrap();
let guest = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
.unwrap()
.unwrap();
guest.make_durable();
backdate_owner();
DrainRequest::new(4, old).write(root.path()).unwrap();
assert!(retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
assert!(root.path().join(FENCE_DIRECTORY).exists());
assert!(!root.path().join(REQUEST_FILE).exists());
}
#[test]
fn heartbeat_and_status_documents_replace_atomically() {
let root = tempfile::tempdir().unwrap();
let first = GuestHeartbeat {
schema_version: SCHEMA_VERSION,
observed_at: Utc::now(),
acknowledged_generation: None,
local_active_attempts: Some(1),
local_busy_attempts: Some(1),
managed_targets: Vec::new(),
unmanaged_runner_services: Some(0),
};
let mut second = first.clone();
second.local_active_attempts = Some(0);
first.write(root.path()).unwrap();
second.write(root.path()).unwrap();
assert_eq!(GuestHeartbeat::read(root.path()).unwrap(), Some(second));
let first = RecoveryStatus {
schema_version: SCHEMA_VERSION,
observed_at: Utc::now(),
phase: RecoveryPhase::Degraded,
consecutive_probe_failures: 1,
reason: None,
last_recovered_at: None,
};
let mut second = first.clone();
second.phase = RecoveryPhase::Healthy;
first.write(root.path()).unwrap();
second.write(root.path()).unwrap();
assert_eq!(RecoveryStatus::read(root.path()).unwrap(), Some(second));
}
}