use std::{
fs,
io::{self, Write},
path::{Path, PathBuf},
process::Stdio,
time::{SystemTime, UNIX_EPOCH},
};
use serde::{Deserialize, Serialize};
use thiserror::Error;
use crate::{reviewer::ReviewQueue, time::unix_now};
pub const WATCHER_LOCK_FILE: &str = "watcher.lock";
const WATCHER_RECLAIM_DIR: &str = "watcher.lock.reclaim";
const LOCK_SETTLING_SECS: u64 = 10;
#[derive(Debug, Error)]
pub enum WatcherError {
#[error("watcher lock io error: {0}")]
Io(io::Error),
#[error("watcher lock json error: {0}")]
Json(serde_json::Error),
#[error("failed to resolve the truth-mirror executable path: {0}")]
ResolveExe(io::Error),
#[error("failed to spawn detached watcher: {0}")]
Spawn(io::Error),
#[error("failed to read review queue: {0}")]
Queue(crate::reviewer::ReviewerError),
}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct ProcessIdentity {
pub pid: u32,
#[serde(default)]
pub start_token: String,
}
impl ProcessIdentity {
pub fn current() -> Self {
let pid = std::process::id();
Self {
start_token: probe_start_token(pid).unwrap_or_default(),
pid,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct WatcherLock {
pub identity: ProcessIdentity,
pub created_at_unix: u64,
}
impl WatcherLock {
fn for_current() -> Self {
Self {
identity: ProcessIdentity::current(),
created_at_unix: unix_now(),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Liveness {
Alive,
Stale,
}
pub fn decide_liveness(recorded: &ProcessIdentity, probed: Option<ProcessIdentity>) -> Liveness {
let Some(probed) = probed else {
return Liveness::Stale;
};
if probed.pid != recorded.pid {
return Liveness::Stale;
}
if recorded.start_token.is_empty() || probed.start_token.is_empty() {
return Liveness::Alive;
}
if probed.start_token == recorded.start_token {
Liveness::Alive
} else {
Liveness::Stale
}
}
fn lock_is_held_at(lock: &WatcherLock, probed: Option<ProcessIdentity>, now: u64) -> bool {
if decide_liveness(&lock.identity, probed) == Liveness::Alive {
return true;
}
now.saturating_sub(lock.created_at_unix) < LOCK_SETTLING_SECS
}
fn lock_is_held(lock: &WatcherLock) -> bool {
let probed = probe_identity(lock.identity.pid);
lock_is_held_at(lock, probed, unix_now())
}
fn probe_identity(pid: u32) -> Option<ProcessIdentity> {
if !pid_is_alive(pid) {
return None;
}
Some(ProcessIdentity {
pid,
start_token: probe_start_token(pid).unwrap_or_default(),
})
}
pub fn watcher_is_alive(state_dir: &Path) -> bool {
match read_lock(state_dir) {
Ok(Some(lock)) => lock_is_held(&lock),
_ => false,
}
}
fn lock_path(state_dir: &Path) -> PathBuf {
state_dir.join(WATCHER_LOCK_FILE)
}
#[derive(Clone, Debug, Eq, PartialEq)]
enum LockRead {
Missing,
Valid(WatcherLock),
Corrupt { modified_at_unix: u64 },
}
impl LockRead {
fn is_held_at(&self, now: u64) -> bool {
match self {
LockRead::Missing => false,
LockRead::Valid(lock) => lock_is_held_at(lock, probe_identity(lock.identity.pid), now),
LockRead::Corrupt { modified_at_unix } => {
corrupt_lock_is_held_at(*modified_at_unix, now)
}
}
}
}
fn corrupt_lock_is_held_at(modified_at_unix: u64, now: u64) -> bool {
now.saturating_sub(modified_at_unix) < LOCK_SETTLING_SECS
}
fn read_lock(state_dir: &Path) -> Result<Option<WatcherLock>, WatcherError> {
match read_lock_state(state_dir)? {
LockRead::Valid(lock) => Ok(Some(lock)),
LockRead::Missing | LockRead::Corrupt { .. } => Ok(None),
}
}
fn read_lock_state(state_dir: &Path) -> Result<LockRead, WatcherError> {
let path = lock_path(state_dir);
match fs::read_to_string(&path) {
Ok(contents) => match serde_json::from_str(contents.trim()) {
Ok(lock) => Ok(LockRead::Valid(lock)),
Err(_) => Ok(LockRead::Corrupt {
modified_at_unix: file_modified_at_unix(&path),
}),
},
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(LockRead::Missing),
Err(error) => Err(WatcherError::Io(error)),
}
}
fn file_modified_at_unix(path: &Path) -> u64 {
fs::metadata(path)
.and_then(|metadata| metadata.modified())
.ok()
.and_then(|modified| modified.duration_since(UNIX_EPOCH).ok())
.map_or_else(unix_now, |duration| duration.as_secs())
}
fn write_lock(state_dir: &Path, lock: &WatcherLock) -> Result<(), WatcherError> {
fs::create_dir_all(state_dir).map_err(WatcherError::Io)?;
let bytes = serde_json::to_vec_pretty(lock).map_err(WatcherError::Json)?;
let temp_path = temp_lock_path(state_dir);
{
let mut file = fs::File::create(&temp_path).map_err(WatcherError::Io)?;
file.write_all(&bytes).map_err(WatcherError::Io)?;
file.write_all(b"\n").map_err(WatcherError::Io)?;
file.sync_all().map_err(WatcherError::Io)?;
}
fs::rename(&temp_path, lock_path(state_dir)).map_err(WatcherError::Io)
}
fn temp_lock_path(state_dir: &Path) -> PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_nanos());
state_dir.join(format!(
"{WATCHER_LOCK_FILE}.{}.{}.tmp",
std::process::id(),
nanos
))
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LockClaim {
Acquired,
HeldByLiveWatcher,
}
pub fn try_acquire_lock(state_dir: &Path) -> Result<LockClaim, WatcherError> {
fs::create_dir_all(state_dir).map_err(WatcherError::Io)?;
let lock = WatcherLock::for_current();
let bytes = serde_json::to_vec_pretty(&lock).map_err(WatcherError::Json)?;
match fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(lock_path(state_dir))
{
Ok(mut file) => {
file.write_all(&bytes).map_err(WatcherError::Io)?;
file.write_all(b"\n").map_err(WatcherError::Io)?;
file.sync_all().map_err(WatcherError::Io)?;
Ok(LockClaim::Acquired)
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
reclaim_existing_lock(state_dir, &lock)
}
Err(error) => Err(WatcherError::Io(error)),
}
}
fn reclaim_existing_lock(state_dir: &Path, lock: &WatcherLock) -> Result<LockClaim, WatcherError> {
let now = unix_now();
let state = read_lock_state(state_dir)?;
if state.is_held_at(now) {
return Ok(LockClaim::HeldByLiveWatcher);
}
let Some(_guard) = try_reclaim_guard(state_dir)? else {
return Ok(LockClaim::HeldByLiveWatcher);
};
let state = read_lock_state(state_dir)?;
if state.is_held_at(unix_now()) {
return Ok(LockClaim::HeldByLiveWatcher);
}
match fs::remove_file(lock_path(state_dir)) {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => return Err(WatcherError::Io(error)),
}
match fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(lock_path(state_dir))
{
Ok(mut file) => {
let bytes = serde_json::to_vec_pretty(lock).map_err(WatcherError::Json)?;
file.write_all(&bytes).map_err(WatcherError::Io)?;
file.write_all(b"\n").map_err(WatcherError::Io)?;
file.sync_all().map_err(WatcherError::Io)?;
Ok(LockClaim::Acquired)
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
Ok(LockClaim::HeldByLiveWatcher)
}
Err(error) => Err(WatcherError::Io(error)),
}
}
struct ReclaimGuard {
path: PathBuf,
}
impl Drop for ReclaimGuard {
fn drop(&mut self) {
let _ = fs::remove_dir(&self.path);
}
}
fn try_reclaim_guard(state_dir: &Path) -> Result<Option<ReclaimGuard>, WatcherError> {
let path = state_dir.join(WATCHER_RECLAIM_DIR);
match fs::create_dir(&path) {
Ok(()) => Ok(Some(ReclaimGuard { path })),
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
if corrupt_lock_is_held_at(file_modified_at_unix(&path), unix_now()) {
return Ok(None);
}
match fs::remove_dir(&path) {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => return Err(WatcherError::Io(error)),
}
match fs::create_dir(&path) {
Ok(()) => Ok(Some(ReclaimGuard { path })),
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => Ok(None),
Err(error) => Err(WatcherError::Io(error)),
}
}
Err(error) => Err(WatcherError::Io(error)),
}
}
pub fn release_lock_if_owned(state_dir: &Path) -> Result<(), WatcherError> {
let Some(lock) = read_lock(state_dir)? else {
return Ok(());
};
if !identity_matches_owner(&lock.identity, &ProcessIdentity::current()) {
return Ok(());
}
match fs::remove_file(lock_path(state_dir)) {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => return Err(WatcherError::Io(error)),
}
Ok(())
}
fn identity_matches_owner(recorded: &ProcessIdentity, current: &ProcessIdentity) -> bool {
recorded.pid == current.pid
&& (recorded.start_token.is_empty()
|| current.start_token.is_empty()
|| recorded.start_token == current.start_token)
}
pub fn ensure_watcher(state_dir: &Path) -> Result<LockClaim, WatcherError> {
ensure_watcher_with_spawner(state_dir, spawn_detached_watcher)
}
fn ensure_watcher_with_spawner(
state_dir: &Path,
spawn: impl FnOnce(&Path) -> Result<u32, WatcherError>,
) -> Result<LockClaim, WatcherError> {
if watcher_is_alive(state_dir) {
return Ok(LockClaim::HeldByLiveWatcher);
}
match try_acquire_lock(state_dir)? {
LockClaim::HeldByLiveWatcher => Ok(LockClaim::HeldByLiveWatcher),
LockClaim::Acquired => {
let child_pid = match spawn(state_dir) {
Ok(pid) => pid,
Err(error) => {
if let Err(release_error) = release_lock_if_owned(state_dir) {
eprintln!(
"truth-mirror ensure-watcher: failed to release lock after spawn failure: {release_error}"
);
}
return Err(error);
}
};
let child_lock = WatcherLock {
identity: ProcessIdentity {
start_token: probe_start_token(child_pid).unwrap_or_default(),
pid: child_pid,
},
created_at_unix: unix_now(),
};
write_lock(state_dir, &child_lock)?;
Ok(LockClaim::Acquired)
}
}
}
fn spawn_detached_watcher(state_dir: &Path) -> Result<u32, WatcherError> {
let exe = std::env::current_exe().map_err(WatcherError::ResolveExe)?;
let mut command = std::process::Command::new(exe);
command
.arg("--state-dir")
.arg(state_dir)
.arg("watch")
.arg("--until-empty")
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
detach(&mut command);
let child = command.spawn().map_err(WatcherError::Spawn)?;
Ok(child.id())
}
#[cfg(unix)]
fn detach(command: &mut std::process::Command) {
use std::os::unix::process::CommandExt as _;
command.process_group(0);
}
#[cfg(windows)]
fn detach(command: &mut std::process::Command) {
use std::os::windows::process::CommandExt as _;
const DETACHED_PROCESS: u32 = 0x0000_0008;
const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
command.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW);
}
#[cfg(not(any(unix, windows)))]
fn detach(_command: &mut std::process::Command) {}
#[cfg(unix)]
pub fn pid_is_alive(pid: u32) -> bool {
std::process::Command::new("kill")
.arg("-0")
.arg(pid.to_string())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map(|status| status.success())
.unwrap_or(false)
}
#[cfg(windows)]
pub fn pid_is_alive(pid: u32) -> bool {
let output = std::process::Command::new("tasklist")
.args(["/FI", &format!("PID eq {pid}"), "/NH", "/FO", "CSV"])
.stderr(Stdio::null())
.output();
match output {
Ok(output) if output.status.success() => {
let text = String::from_utf8_lossy(&output.stdout);
text.contains(&pid.to_string())
}
_ => false,
}
}
#[cfg(not(any(unix, windows)))]
pub fn pid_is_alive(_pid: u32) -> bool {
false
}
#[cfg(unix)]
fn probe_start_token(pid: u32) -> Option<String> {
let output = std::process::Command::new("ps")
.args(["-o", "lstart=", "-p", &pid.to_string()])
.stderr(Stdio::null())
.output()
.ok()?;
if !output.status.success() {
return None;
}
let token = String::from_utf8_lossy(&output.stdout).trim().to_owned();
(!token.is_empty()).then_some(token)
}
#[cfg(windows)]
fn probe_start_token(pid: u32) -> Option<String> {
let script = format!(
"$p = Get-CimInstance Win32_Process -Filter 'ProcessId = {pid}' -ErrorAction Stop; \
if ($null -ne $p) {{ $p.CreationDate.ToUniversalTime().ToString('o') }}"
);
let output = std::process::Command::new("powershell")
.args(["-NoProfile", "-Command", &script])
.stderr(Stdio::null())
.output()
.ok()?;
if !output.status.success() {
return None;
}
parse_powershell_start_token_output(&output.stdout)
}
#[cfg(any(windows, test))]
fn parse_powershell_start_token_output(stdout: &[u8]) -> Option<String> {
let text = String::from_utf8_lossy(stdout);
let token = text
.lines()
.find_map(|line| {
let trimmed = line.trim();
(!trimmed.is_empty()).then_some(trimmed)
})
.map(str::trim)
.map(str::to_owned)?;
(!token.is_empty()).then_some(token)
}
#[cfg(not(any(unix, windows)))]
fn probe_start_token(_pid: u32) -> Option<String> {
None
}
pub fn live_watcher_pid(state_dir: &Path) -> Option<u32> {
match read_lock(state_dir) {
Ok(Some(lock)) => lock_is_held(&lock).then_some(lock.identity.pid),
_ => None,
}
}
pub fn queue_has_pending(state_dir: &Path) -> Result<bool, WatcherError> {
let summary = ReviewQueue::new(state_dir)
.summary()
.map_err(WatcherError::Queue)?;
Ok(summary.pending_count > 0)
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{
Arc, Barrier,
atomic::{AtomicUsize, Ordering},
};
fn identity(pid: u32, token: &str) -> ProcessIdentity {
ProcessIdentity {
pid,
start_token: token.to_owned(),
}
}
#[test]
fn dead_pid_is_stale() {
let recorded = identity(4242, "Mon Jan 1 00:00:00 2026");
assert_eq!(decide_liveness(&recorded, None), Liveness::Stale);
}
#[test]
fn reused_pid_with_different_start_token_is_stale() {
let recorded = identity(4242, "Mon Jan 1 00:00:00 2026");
let probed = identity(4242, "Tue Feb 2 11:11:11 2026");
assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Stale);
}
#[test]
fn same_pid_and_start_token_is_alive() {
let recorded = identity(4242, "Mon Jan 1 00:00:00 2026");
let probed = recorded.clone();
assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Alive);
}
#[test]
fn empty_start_token_falls_back_to_pid_liveness() {
let recorded = identity(4242, "");
let probed = identity(4242, "some-token");
assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Alive);
let recorded = identity(4242, "recorded-token");
let probed = identity(4242, "");
assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Alive);
}
#[test]
fn different_pid_is_stale_even_with_matching_token() {
let recorded = identity(4242, "same-token");
let probed = identity(9999, "same-token");
assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Stale);
}
#[test]
fn acquire_then_reacquire_by_live_current_process_stands_down() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
assert_eq!(
try_acquire_lock(state).unwrap(),
LockClaim::HeldByLiveWatcher
);
assert!(watcher_is_alive(state));
assert_eq!(live_watcher_pid(state), Some(std::process::id()));
}
#[test]
fn empty_or_corrupt_lock_is_held_only_during_settling_window() {
assert!(corrupt_lock_is_held_at(1_000, 1_000));
assert!(corrupt_lock_is_held_at(
1_000,
1_000 + LOCK_SETTLING_SECS - 1
));
assert!(!corrupt_lock_is_held_at(1_000, 1_000 + LOCK_SETTLING_SECS));
}
#[test]
fn empty_lock_file_does_not_error_or_get_reclaimed_while_settling() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
fs::create_dir_all(state).unwrap();
fs::write(lock_path(state), "").unwrap();
assert!(matches!(
read_lock_state(state).unwrap(),
LockRead::Corrupt { .. }
));
assert_eq!(
try_acquire_lock(state).unwrap(),
LockClaim::HeldByLiveWatcher
);
}
#[test]
fn stale_lock_from_dead_pid_is_reclaimed() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
let stale = WatcherLock {
identity: identity(4_000_000_000, "definitely-not-a-live-token"),
created_at_unix: 1,
};
write_lock(state, &stale).unwrap();
assert!(!watcher_is_alive(state));
assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
assert_eq!(live_watcher_pid(state), Some(std::process::id()));
}
#[test]
fn stale_lock_with_unknown_start_token_from_dead_pid_is_reclaimed() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
let stale = WatcherLock {
identity: identity(4_000_000_000, ""),
created_at_unix: 1,
};
write_lock(state, &stale).unwrap();
assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
assert_eq!(live_watcher_pid(state), Some(std::process::id()));
}
#[test]
fn threaded_stale_lock_reclaim_race_yields_exactly_one_winner() {
const THREADS: usize = 8;
let temp = tempfile::tempdir().unwrap();
let state = Arc::new(temp.path().to_path_buf());
let stale = WatcherLock {
identity: identity(4_000_000_000, "definitely-not-a-live-token"),
created_at_unix: 1,
};
write_lock(state.as_ref(), &stale).unwrap();
let barrier = Arc::new(Barrier::new(THREADS));
let acquired = Arc::new(AtomicUsize::new(0));
let held = Arc::new(AtomicUsize::new(0));
let mut threads = Vec::with_capacity(THREADS);
for _ in 0..THREADS {
let state = Arc::clone(&state);
let barrier = Arc::clone(&barrier);
let acquired = Arc::clone(&acquired);
let held = Arc::clone(&held);
threads.push(std::thread::spawn(move || {
barrier.wait();
match try_acquire_lock(state.as_ref()).unwrap() {
LockClaim::Acquired => {
acquired.fetch_add(1, Ordering::SeqCst);
}
LockClaim::HeldByLiveWatcher => {
held.fetch_add(1, Ordering::SeqCst);
}
}
}));
}
for thread in threads {
thread.join().unwrap();
}
assert_eq!(acquired.load(Ordering::SeqCst), 1);
assert_eq!(held.load(Ordering::SeqCst), THREADS - 1);
}
#[test]
fn release_is_a_noop_when_not_owner() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
let other = WatcherLock {
identity: identity(4_000_000_000, "other-token"),
created_at_unix: 1,
};
write_lock(state, &other).unwrap();
release_lock_if_owned(state).unwrap();
assert!(read_lock(state).unwrap().is_some());
}
#[test]
fn release_removes_lock_when_owner() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
release_lock_if_owned(state).unwrap();
assert!(read_lock(state).unwrap().is_none());
}
#[test]
fn release_removes_lock_when_owner_start_token_is_unknown() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
let current = WatcherLock {
identity: identity(std::process::id(), ""),
created_at_unix: 1,
};
write_lock(state, ¤t).unwrap();
release_lock_if_owned(state).unwrap();
assert!(read_lock(state).unwrap().is_none());
}
#[test]
fn ensure_watcher_releases_claimed_lock_when_spawn_fails() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
let error = ensure_watcher_with_spawner(state, |_| {
Err(WatcherError::Spawn(io::Error::other("spawn failed")))
})
.unwrap_err();
assert!(matches!(error, WatcherError::Spawn(_)));
assert!(read_lock(state).unwrap().is_none());
assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
}
#[test]
fn powershell_start_token_parser_uses_first_nonempty_line() {
let output = b"\r\n 2026-07-09T01:02:03.0000000Z \r\n";
assert_eq!(
parse_powershell_start_token_output(output).as_deref(),
Some("2026-07-09T01:02:03.0000000Z")
);
}
#[test]
fn powershell_start_token_parser_rejects_empty_output() {
assert_eq!(parse_powershell_start_token_output(b"\r\n \n"), None);
}
#[test]
fn current_identity_has_our_pid() {
assert_eq!(ProcessIdentity::current().pid, std::process::id());
}
#[test]
fn fresh_lock_with_dead_owner_is_held_during_settling_window() {
let lock = WatcherLock {
identity: identity(4_000_000_000, "dead-token"),
created_at_unix: 1_000,
};
assert!(lock_is_held_at(&lock, None, 1_000));
assert!(lock_is_held_at(&lock, None, 1_000 + LOCK_SETTLING_SECS - 1));
}
#[test]
fn stale_lock_past_settling_window_is_not_held() {
let lock = WatcherLock {
identity: identity(4_000_000_000, "dead-token"),
created_at_unix: 1_000,
};
assert!(!lock_is_held_at(&lock, None, 1_000 + LOCK_SETTLING_SECS));
assert!(!lock_is_held_at(&lock, None, 5_000));
}
#[test]
fn live_owner_is_held_regardless_of_lock_age() {
let lock = WatcherLock {
identity: identity(4242, "live-token"),
created_at_unix: 1,
};
let probed = identity(4242, "live-token");
assert!(lock_is_held_at(&lock, Some(probed), 9_999_999));
}
}