hayson-engine 0.25.0

use std::fmt;
use std::fs::{self, OpenOptions};
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use rand::Rng;
use serde::{Deserialize, Serialize};

use super::platform;

pub const LOCK_FILE: &str = ".hayson.lock";

const DEFAULT_STALE_MS: u64 = 30_000;
const DEFAULT_HEARTBEAT_MS: u64 = 5_000;
const DEFAULT_TIMEOUT_MS: u64 = 5_000;
const DEFAULT_RETRY_MS: u64 = 50;
const MAX_IMMEDIATE_RETRIES: u32 = 2;
/// How finely the heartbeat thread wakes, so that a `release` never waits out a whole beat.
#[cfg(unix)]
const HEARTBEAT_STEP_MS: u64 = 5;
const TEMP_NAME_ATTEMPTS: u32 = 8;
const OWNER_BYTES: usize = 16;

fn now_ms() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// A fresh 128-bit token per `DirLock`. Ownership is compared on this and nothing else, so the
/// protocol needs no pid — which is what lets it run where pids do not exist.
fn mint_owner() -> String {
    use std::fmt::Write as _;
    let mut bytes = [0u8; OWNER_BYTES];
    rand::rng().fill_bytes(&mut bytes);
    let mut owner = String::with_capacity(OWNER_BYTES * 2);
    for byte in bytes {
        let _ = write!(owner, "{byte:02x}");
    }
    owner
}

/// How the writer lock on a database directory is taken.
///
/// `timeout_ms` bounds how long `acquire` waits for another writer to leave before it gives up
/// with [`LockedError`]; `0` fails on the first refusal. `retry_interval_ms` is the polling step
/// while waiting. `stale_after_ms` is the heartbeat age after which a lock is treated as abandoned
/// and reclaimed; `heartbeat_ms` is how often the holder refreshes its own heartbeat.
///
/// Reclaiming an expired lease costs a further `stale_after_ms / 2` of grace on top, and is only
/// attempted when `timeout_ms` leaves room for it — so a timeout shorter than that grace never
/// reclaims, it just reports the directory as locked. `grace_served` is how a caller that waited
/// that grace out on its own clock skips the wait `acquire` would otherwise do inside itself.
#[derive(Debug, Clone, Copy, Default)]
pub struct LockOptions {
    pub stale_after_ms: Option<u64>,
    pub heartbeat_ms: Option<u64>,
    pub timeout_ms: Option<u64>,
    pub retry_interval_ms: Option<u64>,
    /// A lease this caller already challenged and waited out, so the first pass that finds that
    /// very lease still expired reclaims it without sleeping. Spent on that pass whether the
    /// reclaim lands or not; a token naming any other lease is ignored and the ordinary,
    /// sleeping hand-over runs instead.
    pub grace_served: Option<GraceServed>,
}

/// A lease a caller has already challenged and waited the hand-over grace out on, in the shape
/// that survives a trip out to a host and back: the owner token as the number its hex spells, and
/// the heartbeat that token had not moved past.
///
/// It is a claim, not a proof. [`DirLock::acquire`] takes the caller's word that
/// the `grace_ms` of a [`ChallengeOutcome::GraceOwed`] really elapsed and only re-checks that the lease is
/// still the same one — a caller that skips its own wait reclaims a lock its holder was owed
/// notice on. What it can never do is reclaim a *different* lease, or two callers reclaim one.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GraceServed {
    owner: u128,
    heartbeat_at: i64,
}

impl GraceServed {
    /// The token for the holder a [`ChallengeOutcome::GraceOwed`] named.
    ///
    /// @param holder the holder that outcome carried, unchanged
    /// @returns the token, or `None` for an `owner` this engine did not mint — a lock file nobody
    /// can name is never reclaimed on a caller's word
    pub fn of(holder: &LockHolder) -> Option<GraceServed> {
        GraceServed::parse(&holder.owner, holder.heartbeat_at)
    }

    /// As [`GraceServed::of`], from the two members alone, for a host that carries them back out
    /// of its own refusal envelope rather than holding a [`LockHolder`].
    ///
    /// @param owner the holder's `owner` token, as the lock file spells it
    /// @param heartbeat_at the holder's `heartbeat_at`, unchanged
    /// @returns the token, or `None` when `owner` is not one this engine minted
    pub fn parse(owner: &str, heartbeat_at: i64) -> Option<GraceServed> {
        if owner.len() != OWNER_BYTES * 2 {
            return None;
        }
        u128::from_str_radix(owner, 16)
            .ok()
            .map(|owner| GraceServed { owner, heartbeat_at })
    }

    fn names(&self, holder: &LockHolder) -> bool {
        GraceServed::of(holder) == Some(*self)
    }
}

/// The contents of `.hayson.lock`. `owner` is the only field ownership is decided on; `pid` and
/// `host` are diagnostics, present on unix and absent on platforms without processes.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct LockHolder {
    pub owner: String,
    #[serde(default)]
    pub pid: Option<u32>,
    #[serde(default)]
    pub host: Option<String>,
    pub started_at: i64,
    pub heartbeat_at: i64,
}

fn new_holder(owner: &str) -> LockHolder {
    let now = now_ms();
    LockHolder {
        owner: owner.to_string(),
        pid: platform::pid(),
        host: platform::hostname(),
        started_at: now,
        heartbeat_at: now,
    }
}

#[derive(Debug, Clone)]
pub struct LockedError {
    pub subject: PathBuf,
    pub holder: Option<LockHolder>,
    pub waited_ms: u64,
}

impl fmt::Display for LockedError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let by = match &self.holder {
            Some(holder) => match (holder.pid, holder.host.as_deref()) {
                (Some(pid), Some(host)) => format!("pid {pid} on {host}"),
                (Some(pid), None) => format!("pid {pid}"),
                (None, Some(host)) => format!("another writer on {host}"),
                (None, None) => "another writer".to_string(),
            },
            None => "an unknown process".to_string(),
        };
        let waited = if self.waited_ms > 0 {
            format!(" after waiting {} ms", self.waited_ms)
        } else {
            String::new()
        };
        write!(
            f,
            "database \"{}\" is locked for writing by {by}{waited} — raise LockOptions::timeout_ms to wait longer, or open it read-only",
            self.subject.display(),
        )
    }
}

impl std::error::Error for LockedError {}

#[derive(Debug)]
pub enum DirLockError {
    Locked(LockedError),
    /// The lock was reclaimed by somebody else while this handle still believed it held it. Every
    /// write through the handle has to stop: the directory now answers to another writer.
    Lost(PathBuf),
    /// Something other than a regular file sits at the lock path. Reading or replacing it could
    /// write through a link the caller did not choose, so nothing is touched at all.
    SuspiciousLockFile(PathBuf),
    Io(io::Error),
}

impl fmt::Display for DirLockError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            DirLockError::Locked(error) => error.fmt(f),
            DirLockError::Lost(path) => write!(
                f,
                "the writer lock \"{}\" was taken over by another writer — this handle no longer holds it and must not write; reopen the database",
                path.display(),
            ),
            DirLockError::SuspiciousLockFile(path) => write!(
                f,
                "the lock file \"{}\" is a symbolic link, not a regular file — refusing to read or replace it; remove whatever planted it",
                path.display(),
            ),
            DirLockError::Io(error) => error.fmt(f),
        }
    }
}

impl std::error::Error for DirLockError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            DirLockError::Locked(error) => Some(error),
            DirLockError::Io(error) => Some(error),
            DirLockError::Lost(_) | DirLockError::SuspiciousLockFile(_) => None,
        }
    }
}

impl From<io::Error> for DirLockError {
    fn from(error: io::Error) -> Self {
        DirLockError::Io(error)
    }
}

enum LockFileState {
    Holder(LockHolder),
    Missing,
    Unreadable { touched_at_ms: i64 },
}

/// Reads `.hayson.lock` after establishing that it really is a regular file. A lock file that
/// parsed under an older format lands in `Unreadable`, which is the reclaim-after-`stale_after_ms`
/// path — the only migration `.hayson.lock` needs, since it never survives a clean shutdown.
fn read_lock_file(path: &Path) -> Result<LockFileState, DirLockError> {
    let metadata = match fs::symlink_metadata(path) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(LockFileState::Missing),
        Err(_) => {
            return Ok(LockFileState::Unreadable {
                touched_at_ms: now_ms(),
            })
        }
    };
    if metadata.file_type().is_symlink() {
        return Err(DirLockError::SuspiciousLockFile(path.to_path_buf()));
    }
    let touched_at_ms = metadata
        .modified()
        .ok()
        .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
        .map(|d| d.as_millis() as i64)
        .unwrap_or_else(now_ms);

    let raw = match fs::read_to_string(path) {
        Ok(raw) => raw,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(LockFileState::Missing),
        Err(_) => return Ok(LockFileState::Unreadable { touched_at_ms }),
    };
    match serde_json::from_str::<LockHolder>(&raw) {
        Ok(holder) => Ok(LockFileState::Holder(holder)),
        Err(_) => Ok(LockFileState::Unreadable { touched_at_ms }),
    }
}

enum Abandonment {
    Held,
    /// Nobody renewed the lease in time. Reclaiming needs the challenge in [`steal`] first.
    LeaseExpired,
    /// The holding process is gone, on this host, right now — there is nothing left that could
    /// renew the lease, so the challenge would only cost latency. Unix only.
    ProvablyDead,
}

fn provably_dead(holder: &LockHolder) -> bool {
    let (Some(pid), Some(host)) = (holder.pid, holder.host.as_deref()) else {
        return false;
    };
    if platform::hostname().as_deref() != Some(host) {
        return false;
    }
    platform::process_alive(pid) == Some(false)
}

fn abandonment(holder: &LockHolder, stale_after_ms: u64) -> Abandonment {
    if provably_dead(holder) {
        return Abandonment::ProvablyDead;
    }
    if now_ms() - holder.heartbeat_at > stale_after_ms as i64 {
        return Abandonment::LeaseExpired;
    }
    Abandonment::Held
}

fn temp_path_for(path: &Path, owner: &str, ordinal: u64) -> PathBuf {
    PathBuf::from(format!("{}.{owner}.{ordinal}.tmp", path.display()))
}

/// Serializes `holder` into a temp file nobody else can be holding: the name carries this lock's
/// own owner token, and `create_new` refuses anything already at it — a regular file, and a
/// planted symlink alike — so the victim of a planted link is never written through. Returns the
/// temp path, ready to be renamed over the lock; the caller owns it from there.
fn write_temp(path: &Path, owner: &str, counter: &AtomicU64, holder: &LockHolder) -> io::Result<PathBuf> {
    let json = serde_json::to_string(holder)?;
    let mut refusal = None;
    for _ in 0..TEMP_NAME_ATTEMPTS {
        let temp = temp_path_for(path, owner, counter.fetch_add(1, Ordering::SeqCst));
        let mut file = match OpenOptions::new().write(true).create_new(true).open(&temp) {
            Ok(file) => file,
            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
                refusal = Some(error);
                continue;
            }
            Err(error) => return Err(error),
        };
        let written = file.write_all(json.as_bytes()).and_then(|()| file.sync_all());
        drop(file);
        if let Err(error) = written {
            let _ = fs::remove_file(&temp);
            return Err(error);
        }
        return Ok(temp);
    }
    Err(refusal.unwrap_or_else(|| {
        io::Error::new(
            io::ErrorKind::AlreadyExists,
            "every candidate lock temporary was already taken",
        )
    }))
}

fn still_matches(path: &Path, observed: &LockHolder) -> Result<bool, DirLockError> {
    Ok(match read_lock_file(path)? {
        LockFileState::Holder(seen) => seen.owner == observed.owner && seen.heartbeat_at == observed.heartbeat_at,
        LockFileState::Missing | LockFileState::Unreadable { .. } => false,
    })
}

/// Takes an abandoned lock away from whoever wrote it, in four steps that each get a chance to
/// call the whole thing off:
///
/// 1. write this challenger's own holder to a temp file nothing else can be at;
/// 2. re-read the lock file, and abandon the steal if anything moved in the meantime;
/// 3. rename the temp over the lock file, which replaces it atomically and — unlike an open for
///    writing — does not follow a symlink sitting at the destination;
/// 4. read the lock file back, and keep the lock only if it names this challenger.
///
/// Step 2 narrows the window between the last check and the rename; it does not close it, so two
/// challengers can both reach the rename and the later one wins. Step 4 is what makes that
/// survivable: the loser sees somebody else's token and goes back to waiting rather than believing
/// it holds a lock it does not. A rename that lands between another challenger's rename and its
/// read-back leaves that one thinking it won for as long as it takes its next `tick` to say
/// otherwise — bounded by `heartbeat_ms`, and fenced when it comes.
///
/// Reached bare only for a holder that is provably dead, which has nothing left to give notice
/// to. Every reclaim that rests on the lease alone goes through [`finish_after_grace`] instead.
fn seize(
    path: &Path,
    observed: &LockHolder,
    owner: &str,
    counter: &AtomicU64,
) -> Result<Option<LockHolder>, DirLockError> {
    let holder = new_holder(owner);
    let temp = write_temp(path, owner, counter, &holder)?;
    match still_matches(path, observed) {
        Ok(true) => {}
        Ok(false) => {
            let _ = fs::remove_file(&temp);
            return Ok(None);
        }
        Err(error) => {
            let _ = fs::remove_file(&temp);
            return Err(error);
        }
    }
    if let Err(error) = fs::rename(&temp, path) {
        let _ = fs::remove_file(&temp);
        return Err(error.into());
    }
    match read_lock_file(path)? {
        LockFileState::Holder(landed) if landed.owner == holder.owner => Ok(Some(holder)),
        _ => Ok(None),
    }
}

/// The far side of a hand-over grace: re-read the lock file and go on to [`seize`] only if the
/// lease is still the very one that was challenged. An owner that changed, or a heartbeat that
/// moved, means somebody is alive there after all and the steal is called off.
///
/// The grace itself belongs to whoever calls this, which is the whole point of the split: where
/// the engine runs on the one thread a host has, a sleep in here is a wait that host pays for.
fn finish_after_grace(
    path: &Path,
    observed: &LockHolder,
    owner: &str,
    counter: &AtomicU64,
) -> Result<Option<LockHolder>, DirLockError> {
    if !still_matches(path, observed)? {
        return Ok(None);
    }
    seize(path, observed, owner, counter)
}

/// What one pass over the lock file found, waiting for nothing on the way. Every answer but
/// [`Pass::Held`] leaves it to the caller to decide whether to wait, and on whose clock.
enum Pass {
    /// The lock is this `owner`'s now — claimed outright, or taken from a holder proven dead.
    Held(LockHolder),
    /// Somebody else holds it. `Some` names the holder; `None` is a lock file too freshly written
    /// to judge — unreadable, and not yet old enough to be treated as abandoned.
    Blocked(Option<LockHolder>),
    /// Nobody renewed the lease. Reclaiming it owes the holder a grace, then
    /// [`finish_after_grace`].
    GraceOwed(LockHolder),
    /// The lock file is gone — released, or removed here as unreadable and long untouched. There
    /// is something to try again at once, with nothing to wait for.
    Vacated,
}

fn one_pass(path: &Path, owner: &str, counter: &AtomicU64, stale_after_ms: u64) -> Result<Pass, DirLockError> {
    if let Some(holder) = claim(path, owner)? {
        return Ok(Pass::Held(holder));
    }
    match read_lock_file(path)? {
        LockFileState::Holder(holder) => match abandonment(&holder, stale_after_ms) {
            Abandonment::Held => Ok(Pass::Blocked(Some(holder))),
            Abandonment::ProvablyDead => match seize(path, &holder, owner, counter)? {
                Some(taken) => Ok(Pass::Held(taken)),
                None => Ok(Pass::Blocked(Some(holder))),
            },
            Abandonment::LeaseExpired => Ok(Pass::GraceOwed(holder)),
        },
        LockFileState::Missing => Ok(Pass::Vacated),
        LockFileState::Unreadable { touched_at_ms } => {
            if now_ms() - touched_at_ms > stale_after_ms as i64 {
                let _ = fs::remove_file(path);
                Ok(Pass::Vacated)
            } else {
                Ok(Pass::Blocked(None))
            }
        }
    }
}

fn claim(path: &Path, owner: &str) -> io::Result<Option<LockHolder>> {
    let mut file = match OpenOptions::new().write(true).create_new(true).open(path) {
        Ok(file) => file,
        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => return Ok(None),
        Err(error) => return Err(error),
    };
    let holder = new_holder(owner);
    let json = serde_json::to_string(&holder)?;
    file.write_all(json.as_bytes())?;
    file.sync_all()?;
    Ok(Some(holder))
}

/// The part of a [`DirLock`] the heartbeat thread also holds, so both can renew and both can see
/// the lock go away underneath them.
#[derive(Debug)]
struct Beat {
    path: PathBuf,
    owner: String,
    heartbeat_ms: u64,
    counter: AtomicU64,
    held: AtomicBool,
    lost: AtomicBool,
    stop: AtomicBool,
    last_beat_ms: AtomicI64,
}

impl Beat {
    fn lose(&self) -> DirLockError {
        self.lost.store(true, Ordering::SeqCst);
        DirLockError::Lost(self.path.clone())
    }

    /// Rewrites the lock file with a current timestamp, having first confirmed it still names this
    /// owner. Finding anybody else's token there — or no lock file at all — is not something to
    /// shrug off: the handle is marked lost for good and every later call says so.
    fn renew(&self) -> Result<(), DirLockError> {
        if self.lost.load(Ordering::SeqCst) {
            return Err(DirLockError::Lost(self.path.clone()));
        }
        if !self.held.load(Ordering::SeqCst) {
            return Ok(());
        }
        let holder = match read_lock_file(&self.path) {
            Ok(LockFileState::Holder(holder)) if holder.owner == self.owner => holder,
            Ok(_) => return Err(self.lose()),
            Err(error) => {
                self.lost.store(true, Ordering::SeqCst);
                return Err(error);
            }
        };
        let beaten = LockHolder {
            heartbeat_at: now_ms(),
            ..holder
        };
        let temp = write_temp(&self.path, &self.owner, &self.counter, &beaten)?;
        if !self.held.load(Ordering::SeqCst) {
            let _ = fs::remove_file(&temp);
            return Ok(());
        }
        if let Err(error) = fs::rename(&temp, &self.path) {
            let _ = fs::remove_file(&temp);
            return Err(error.into());
        }
        self.last_beat_ms.store(now_ms(), Ordering::SeqCst);
        Ok(())
    }

    fn renew_if_due(&self) -> Result<(), DirLockError> {
        if self.lost.load(Ordering::SeqCst) {
            return Err(DirLockError::Lost(self.path.clone()));
        }
        if !self.held.load(Ordering::SeqCst) {
            return Ok(());
        }
        if now_ms() - self.last_beat_ms.load(Ordering::SeqCst) < self.heartbeat_ms as i64 {
            return Ok(());
        }
        self.renew()
    }
}

/// Advisory, single-writer lock on a database directory (ADR-hjxp/0006), claimed with
/// `O_CREAT|O_EXCL` and kept alive by a heartbeat that is written to a fresh temporary and renamed
/// over the lock file (ADR-6wjp/0016), so a stale timestamp is never mistaken for proof that the
/// holder is gone and no write ever follows a link somebody planted in the way.
///
/// The lease — `stale_after_ms` since the last heartbeat — is the whole liveness story, and holds
/// on every platform. Where processes can be asked about (unix), a holder whose pid is gone is
/// reclaimed sooner; that is an accelerator and nothing rests on it.
///
/// The heartbeat runs on a background thread where threads exist. Where they do not, or where one
/// could not be started, [`DirLock::tick`] and the renewal every write already performs are what
/// keeps the lease alive — call `tick` from a host timer at least twice per `heartbeat_ms`.
#[derive(Debug)]
pub struct DirLock {
    beat: Arc<Beat>,
    #[cfg(unix)]
    heartbeat: Option<thread::JoinHandle<()>>,
}

impl DirLock {
    /// `path` is the absolute path of the lock file itself, not of the directory holding it.
    pub fn acquire(path: impl Into<PathBuf>, options: LockOptions) -> Result<DirLock, DirLockError> {
        let path = path.into();
        let subject = path.parent().map(Path::to_path_buf).unwrap_or_else(|| path.clone());
        Self::acquire_as(path, options, subject)
    }

    /// As [`Self::acquire`], but names `subject` — the database root, not the lock file — in a
    /// resulting [`LockedError`].
    pub fn acquire_as(path: PathBuf, options: LockOptions, subject: PathBuf) -> Result<DirLock, DirLockError> {
        let stale_after_ms = options.stale_after_ms.unwrap_or(DEFAULT_STALE_MS);
        let heartbeat_ms = options.heartbeat_ms.unwrap_or(DEFAULT_HEARTBEAT_MS);
        let timeout_ms = options.timeout_ms.unwrap_or(DEFAULT_TIMEOUT_MS);
        let retry_interval_ms = options.retry_interval_ms.unwrap_or(DEFAULT_RETRY_MS).max(1);
        let grace = Duration::from_millis(stale_after_ms / 2);

        let owner = mint_owner();
        let counter = AtomicU64::new(0);
        let started_at = Instant::now();
        let deadline = started_at + Duration::from_millis(timeout_ms);
        let mut retries = 0u32;
        let mut blocker: Option<LockHolder> = None;
        let mut last_seen: Option<LockHolder> = None;
        let mut served = options.grace_served;

        loop {
            let mut retry_now = false;
            match one_pass(&path, &owner, &counter, stale_after_ms)? {
                Pass::Held(holder) => return Ok(DirLock::start(path, owner, counter, heartbeat_ms, &holder)),
                Pass::Blocked(seen) => {
                    if let Some(holder) = seen {
                        last_seen = Some(holder.clone());
                        blocker = Some(holder);
                    }
                }
                Pass::GraceOwed(holder) => {
                    last_seen = Some(holder.clone());
                    // NOTE: the caller's own wait buys exactly one pass, spent whether it lands.
                    let already_served = served.is_some_and(|served| served.names(&holder));
                    if already_served {
                        served = None;
                    }
                    let taken = if already_served {
                        finish_after_grace(&path, &holder, &owner, &counter)?
                    // NOTE: no grace means no reclaim — the caller asked for a shorter answer
                    // than a safe hand-over takes, so report locked instead of overstaying.
                    } else if deadline.saturating_duration_since(Instant::now()) >= grace {
                        thread::sleep(grace);
                        finish_after_grace(&path, &holder, &owner, &counter)?
                    } else {
                        None
                    };
                    if let Some(taken) = taken {
                        return Ok(DirLock::start(path, owner, counter, heartbeat_ms, &taken));
                    }
                    blocker = Some(holder);
                }
                Pass::Vacated => retry_now = true,
            }

            if retry_now && retries < MAX_IMMEDIATE_RETRIES {
                retries += 1;
                continue;
            }

            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Err(DirLockError::Locked(LockedError {
                    subject,
                    holder: blocker.or(last_seen),
                    waited_ms: started_at.elapsed().as_millis() as u64,
                }));
            }
            thread::sleep(remaining.min(Duration::from_millis(retry_interval_ms)));
        }
    }

    /// One pass of the acquire protocol that waits for nothing, so that every wait belongs to the
    /// caller. Where the engine runs on the one thread a host has — `wasm32-wasip1` — this and
    /// [`Self::finish_challenge`] are what let a reclaim happen without the module sleeping
    /// inside a call the host is waiting on.
    ///
    /// `timeout_ms` and `retry_interval_ms` are not read: there is nothing here to time out. A
    /// lock file that vanishes underneath the pass is retried at once, up to the same small bound
    /// [`Self::acquire`] uses, because that costs no waiting either.
    ///
    /// @param path the absolute path of the lock file itself, not of the directory holding it
    /// @param options the lease tuning — `stale_after_ms` and `heartbeat_ms` as `acquire` reads them
    /// @returns the lock, a refusal to reclaim, or the grace a hand-over is owed first
    pub fn challenge(path: impl Into<PathBuf>, options: LockOptions) -> Result<ChallengeOutcome, DirLockError> {
        let path = path.into();
        let stale_after_ms = options.stale_after_ms.unwrap_or(DEFAULT_STALE_MS);
        let heartbeat_ms = options.heartbeat_ms.unwrap_or(DEFAULT_HEARTBEAT_MS);
        let owner = mint_owner();
        let counter = AtomicU64::new(0);

        for _ in 0..=MAX_IMMEDIATE_RETRIES {
            match one_pass(&path, &owner, &counter, stale_after_ms)? {
                Pass::Held(holder) => {
                    let lock = DirLock::start(path, owner, counter, heartbeat_ms, &holder);
                    return Ok(ChallengeOutcome::Taken(lock));
                }
                Pass::Blocked(seen) => return Ok(ChallengeOutcome::Blocked(seen)),
                Pass::GraceOwed(holder) => {
                    return Ok(ChallengeOutcome::GraceOwed {
                        holder,
                        grace_ms: stale_after_ms / 2,
                    })
                }
                Pass::Vacated => {}
            }
        }
        Ok(ChallengeOutcome::Blocked(None))
    }

    /// The far side of a [`ChallengeOutcome::GraceOwed`], for a caller that has waited `grace_ms`
    /// out on its own clock. Reclaims the lock only if the lease is still the one that was
    /// challenged, and does no waiting of its own.
    ///
    /// Two callers finishing the same challenge at the same moment are safe the way two
    /// [`Self::acquire`] calls are: the lock file ends up naming exactly one of them, and any
    /// other that believed it won is fenced the first time it calls [`Self::tick`].
    ///
    /// @param path the absolute path of the lock file itself, not of the directory holding it
    /// @param options the lease tuning — only `heartbeat_ms` is read
    /// @param holder the holder the challenge named, unchanged
    /// @returns the lock, or `None` when somebody proved alive there or reclaimed it first — in
    /// which case the caller is back to [`Self::challenge`]
    pub fn finish_challenge(
        path: impl Into<PathBuf>,
        options: LockOptions,
        holder: &LockHolder,
    ) -> Result<Option<DirLock>, DirLockError> {
        let path = path.into();
        let heartbeat_ms = options.heartbeat_ms.unwrap_or(DEFAULT_HEARTBEAT_MS);
        let owner = mint_owner();
        let counter = AtomicU64::new(0);
        let taken = finish_after_grace(&path, holder, &owner, &counter)?;
        Ok(taken.map(|taken| DirLock::start(path, owner, counter, heartbeat_ms, &taken)))
    }

    /// Reads what the lock file says without claiming, reclaiming or removing anything — the only
    /// entry point here that leaves the directory exactly as it found it, and so the only one an
    /// observer can use against a database somebody else is writing.
    ///
    /// The answer is a snapshot of one moment: a holder reported as [`LockState::Held`] can release
    /// the lock the instant after, and a lease reported as expired can be renewed. It is a
    /// diagnosis, never a reservation — nothing here entitles the caller to the lock.
    ///
    /// @param path the absolute path of the lock file itself, not of the directory holding it
    /// @param options the lease tuning — only `stale_after_ms` is read, to date the lease
    /// @returns what the lock file holds, or [`DirLockError::SuspiciousLockFile`] when something
    /// other than a regular file sits at `path`
    pub fn inspect(path: impl AsRef<Path>, options: LockOptions) -> Result<LockState, DirLockError> {
        let stale_after_ms = options.stale_after_ms.unwrap_or(DEFAULT_STALE_MS);
        Ok(match read_lock_file(path.as_ref())? {
            LockFileState::Missing => LockState::Free,
            LockFileState::Unreadable { .. } => LockState::Unreadable,
            LockFileState::Holder(holder) => match abandonment(&holder, stale_after_ms) {
                Abandonment::Held => LockState::Held(holder),
                Abandonment::LeaseExpired => LockState::LeaseExpired(holder),
                Abandonment::ProvablyDead => LockState::HolderGone(holder),
            },
        })
    }

    fn start(path: PathBuf, owner: String, counter: AtomicU64, heartbeat_ms: u64, holder: &LockHolder) -> DirLock {
        let beat = Arc::new(Beat {
            path,
            owner,
            heartbeat_ms,
            counter,
            held: AtomicBool::new(true),
            lost: AtomicBool::new(false),
            stop: AtomicBool::new(false),
            last_beat_ms: AtomicI64::new(holder.heartbeat_at),
        });

        #[cfg(unix)]
        let heartbeat = spawn_heartbeat(Arc::clone(&beat), heartbeat_ms);

        DirLock {
            beat,
            #[cfg(unix)]
            heartbeat,
        }
    }

    /// Renews the heartbeat if `heartbeat_ms` has elapsed since the last renewal, and does nothing
    /// otherwise — cheap enough to call on every write, and safe to call from a host timer. This
    /// is the only renewal a platform without background threads gets, so a host that drives one
    /// must call it at least twice per `heartbeat_ms`.
    ///
    /// Returns [`DirLockError::Lost`] once the lock has been reclaimed by somebody else; a handle
    /// that has heard that must stop writing, and every later call repeats it.
    pub fn tick(&self) -> Result<(), DirLockError> {
        self.beat.renew_if_due()
    }

    /// Whether this handle still holds the lock as far as it knows — `false` once a `tick` or a
    /// heartbeat has caught somebody else's token in the lock file.
    pub fn is_lost(&self) -> bool {
        self.beat.lost.load(Ordering::SeqCst)
    }

    pub fn release(&mut self) {
        if !self.beat.held.swap(false, Ordering::SeqCst) {
            return;
        }
        self.beat.stop.store(true, Ordering::SeqCst);
        #[cfg(unix)]
        if let Some(handle) = self.heartbeat.take() {
            let _ = handle.join();
        }
        if self.beat.lost.load(Ordering::SeqCst) {
            return;
        }
        // NOTE: only ever remove a lock file that still names this owner — a stolen one is now
        // somebody else's, and removing it would hand the directory to a third writer.
        if let Ok(LockFileState::Holder(holder)) = read_lock_file(&self.beat.path) {
            if holder.owner == self.beat.owner {
                let _ = fs::remove_file(&self.beat.path);
            }
        }
    }

    #[cfg(test)]
    fn owner(&self) -> &str {
        &self.beat.owner
    }

    #[cfg(test)]
    fn temp_path_at(&self, ordinal: u64) -> PathBuf {
        temp_path_for(&self.beat.path, &self.beat.owner, ordinal)
    }
}

/// What [`DirLock::challenge`] found, having waited for nothing to find it.
#[derive(Debug)]
pub enum ChallengeOutcome {
    /// The lock is this handle's — the directory was free, or its holder was provably dead and so
    /// had nothing left to give notice to.
    Taken(DirLock),
    /// Another writer holds it on a lease that is still current, and there is nothing to reclaim.
    /// `Some` names the holder; `None` is a lock file too freshly written to judge.
    Blocked(Option<LockHolder>),
    /// Nobody renewed the lease. A safe hand-over owes the holder `grace_ms` of continued silence
    /// before the lock changes hands: wait that out on a clock that blocks nobody, then call
    /// [`DirLock::finish_challenge`] with this `holder`.
    GraceOwed { holder: LockHolder, grace_ms: u64 },
}

/// What [`DirLock::inspect`] read out of the lock file, having changed nothing to read it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LockState {
    /// There is no lock file: nothing stands between the next writer and the directory.
    Free,
    /// A writer holds the directory on a lease it is still renewing.
    Held(LockHolder),
    /// Nobody renewed the lease within `stale_after_ms`. It is still this holder's until the
    /// hand-over grace [`DirLock::acquire`] owes it has passed.
    LeaseExpired(LockHolder),
    /// The holding process is gone on this host, so nothing is left to renew the lease and the
    /// next writer takes the lock outright. Unix only — elsewhere such a lock reads as `Held`
    /// until its lease runs out.
    HolderGone(LockHolder),
    /// A lock file this engine cannot read: written by an older format, or by something else
    /// entirely. It is reclaimed as abandoned once `stale_after_ms` has passed since it was last
    /// touched, so it blocks a writer meanwhile.
    Unreadable,
}

impl LockState {
    /// The holder the lock file names, where there is one — `None` for a directory nobody holds
    /// and for a lock file that does not parse.
    pub fn holder(&self) -> Option<&LockHolder> {
        match self {
            LockState::Held(holder) | LockState::LeaseExpired(holder) | LockState::HolderGone(holder) => Some(holder),
            LockState::Free | LockState::Unreadable => None,
        }
    }

    /// Whether a writer opening the directory at this moment would have to wait for somebody.
    ///
    /// False for [`LockState::HolderGone`] as well as [`LockState::Free`]: a holder proven dead has
    /// nothing left to give notice to, so `acquire` seizes its lock without waiting at all.
    pub fn blocks_a_writer(&self) -> bool {
        match self {
            LockState::Held(_) | LockState::LeaseExpired(_) | LockState::Unreadable => true,
            LockState::Free | LockState::HolderGone(_) => false,
        }
    }
}

/// NOTE: `Builder::spawn`, not `thread::spawn` — a host out of threads gets tick-only renewal, not
/// a panic. Returns `None` when the thread could not be started.
#[cfg(unix)]
fn spawn_heartbeat(beat: Arc<Beat>, heartbeat_ms: u64) -> Option<thread::JoinHandle<()>> {
    let step = Duration::from_millis(heartbeat_ms.clamp(1, HEARTBEAT_STEP_MS));
    let ticks_per_beat = (heartbeat_ms / step.as_millis().max(1) as u64).max(1);
    thread::Builder::new()
        .name("hayson-lock-heartbeat".to_string())
        .spawn(move || loop {
            for _ in 0..ticks_per_beat {
                if beat.stop.load(Ordering::SeqCst) {
                    return;
                }
                thread::sleep(step);
            }
            if beat.stop.load(Ordering::SeqCst) {
                return;
            }
            // NOTE: only Lost ends the thread — a transient IO failure is retried next beat.
            if matches!(beat.renew(), Err(DirLockError::Lost(_))) {
                return;
            }
        })
        .ok()
}

impl Drop for DirLock {
    fn drop(&mut self) {
        // NOTE: no fs cleanup here on purpose — release() is the only place that removes files.
        self.beat.stop.store(true, Ordering::SeqCst);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lock::platform::tests::{with_pid_liveness, without_pid_liveness};
    use std::os::unix::fs::symlink;

    fn temp_dir(prefix: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!(
            "{prefix}{}-{}",
            std::process::id(),
            SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()
        ));
        fs::create_dir_all(&dir).unwrap();
        dir
    }

    fn holder_of(path: &Path) -> LockHolder {
        serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap()
    }

    fn until<T>(mut seen: impl FnMut() -> Option<T>, within_ms: u64) -> Option<T> {
        let deadline = Instant::now() + Duration::from_millis(within_ms);
        loop {
            if let Some(found) = seen() {
                return Some(found);
            }
            if Instant::now() >= deadline {
                return None;
            }
            thread::sleep(Duration::from_millis(10));
        }
    }

    #[test]
    fn writes_the_holder_and_removes_the_file_on_release() {
        let dir = temp_dir("hayson-lock-basic-");
        let path = dir.join(LOCK_FILE);
        let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();

        let holder = holder_of(&path);
        assert_eq!(holder.owner, lock.owner());
        assert_eq!(holder.pid, Some(std::process::id()));
        assert!(holder.host.is_some());

        lock.release();
        assert!(!path.exists());
    }

    #[test]
    fn refuses_a_second_writer_while_the_first_is_alive() {
        let dir = temp_dir("hayson-lock-contend-");
        let path = dir.join(LOCK_FILE);
        let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();

        let second = DirLock::acquire(
            path.clone(),
            LockOptions {
                timeout_ms: Some(0),
                ..Default::default()
            },
        );
        assert!(matches!(second, Err(DirLockError::Locked(_))));

        lock.release();
        DirLock::acquire(path, LockOptions::default()).unwrap().release();
    }

    #[test]
    fn a_clean_release_frees_the_directory_at_once_without_waiting_out_the_lease() {
        let dir = temp_dir("hayson-lock-clean-release-");
        let path = dir.join(LOCK_FILE);
        let options = LockOptions {
            stale_after_ms: Some(60_000),
            timeout_ms: Some(0),
            ..Default::default()
        };
        let mut first = DirLock::acquire(path.clone(), options).unwrap();
        first.release();

        let started_at = Instant::now();
        let mut second = DirLock::acquire(path, options).unwrap();
        assert!(started_at.elapsed() < Duration::from_millis(500));
        second.release();
    }

    #[test]
    fn waits_up_to_the_timeout_and_takes_the_lock_once_released() {
        let dir = temp_dir("hayson-lock-wait-");
        let path = dir.join(LOCK_FILE);
        let mut first = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();

        thread::spawn(move || {
            thread::sleep(Duration::from_millis(150));
            first.release();
        });

        let started_at = Instant::now();
        let mut second = DirLock::acquire(
            path,
            LockOptions {
                timeout_ms: Some(5_000),
                retry_interval_ms: Some(10),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(started_at.elapsed() >= Duration::from_millis(100));
        second.release();
    }

    #[test]
    fn reports_how_long_it_waited_before_giving_up() {
        let dir = temp_dir("hayson-lock-report-");
        let path = dir.join(LOCK_FILE);
        let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();

        let failure = DirLock::acquire(
            path,
            LockOptions {
                timeout_ms: Some(200),
                retry_interval_ms: Some(10),
                ..Default::default()
            },
        )
        .unwrap_err();
        match failure {
            DirLockError::Locked(error) => {
                assert!(error.waited_ms >= 180);
                assert_eq!(error.holder.unwrap().owner, lock.owner());
            }
            other => panic!("expected Locked, got {other}"),
        }
        lock.release();
    }

    #[test]
    fn reclaims_a_lock_whose_owner_is_gone() {
        // NOTE: this is the pid-liveness accelerator itself — it must not run while another test
        // has the switch turned off, or the lease (60 s here) becomes the only way out.
        let _pids = with_pid_liveness();
        let dir = temp_dir("hayson-lock-dead-owner-");
        let path = dir.join(LOCK_FILE);
        let dead = LockHolder {
            owner: mint_owner(),
            pid: Some(2u32.pow(22)),
            host: platform::hostname(),
            started_at: now_ms(),
            heartbeat_at: now_ms(),
        };
        fs::write(&path, serde_json::to_string(&dead).unwrap()).unwrap();

        let mut lock = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(60_000),
                timeout_ms: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }

    /// The one test that exercises what every non-unix target is limited to: no pid to ask about,
    /// so the lease and the challenge that follows it are the entire liveness story.
    #[test]
    fn reclaims_an_expired_lease_on_the_lease_alone_with_no_pid_liveness_to_help() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-lease-only-");
        let path = dir.join(LOCK_FILE);
        let stale = LockHolder {
            owner: mint_owner(),
            // NOTE: this process's own live pid — a pid check would call it alive and never steal.
            pid: Some(std::process::id()),
            host: platform::hostname(),
            started_at: 0,
            heartbeat_at: 0,
        };
        fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();

        let mut lock = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(120),
                timeout_ms: Some(2_000),
                retry_interval_ms: Some(10),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }

    #[test]
    fn reclaims_a_lock_whose_heartbeat_went_stale() {
        let dir = temp_dir("hayson-lock-stale-heartbeat-");
        let path = dir.join(LOCK_FILE);
        let stale = LockHolder {
            owner: mint_owner(),
            pid: Some(std::process::id()),
            host: Some("some-other-host".to_string()),
            started_at: 0,
            heartbeat_at: 0,
        };
        fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();

        let mut lock = DirLock::acquire(
            path,
            LockOptions {
                stale_after_ms: Some(1),
                ..Default::default()
            },
        )
        .unwrap();
        lock.release();
    }

    #[test]
    fn inspect_names_the_live_holder_and_leaves_the_lock_where_it_is() {
        let dir = temp_dir("hayson-lock-inspect-held-");
        let path = dir.join(LOCK_FILE);
        let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();

        let state = DirLock::inspect(&path, LockOptions::default()).unwrap();

        match &state {
            LockState::Held(holder) => {
                assert_eq!(holder.owner, lock.owner());
                assert_eq!(holder.pid, Some(std::process::id()));
            }
            other => panic!("expected the live holder, got {other:?}"),
        }
        assert!(state.blocks_a_writer());
        assert_eq!(state.holder().map(|holder| holder.owner.as_str()), Some(lock.owner()));
        // The inspected lock is still the first handle's: an observer takes nothing.
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }

    #[test]
    fn inspect_reads_a_directory_nobody_holds_as_free_without_creating_a_lock_file() {
        let dir = temp_dir("hayson-lock-inspect-free-");
        let path = dir.join(LOCK_FILE);

        let state = DirLock::inspect(&path, LockOptions::default()).unwrap();

        assert_eq!(state, LockState::Free);
        assert!(!state.blocks_a_writer());
        assert!(!path.exists(), "inspecting must not plant a lock file");
    }

    #[test]
    fn inspect_tells_an_expired_lease_from_a_holder_that_is_provably_gone() {
        let _pids = with_pid_liveness();
        let dir = temp_dir("hayson-lock-inspect-abandoned-");
        let stale_path = dir.join(LOCK_FILE);
        let dead_path = dir.join("dead.lock");
        let holder = |pid: u32| LockHolder {
            owner: mint_owner(),
            pid: Some(pid),
            host: platform::hostname(),
            started_at: 0,
            heartbeat_at: 0,
        };
        fs::write(&stale_path, serde_json::to_string(&holder(std::process::id())).unwrap()).unwrap();
        fs::write(&dead_path, serde_json::to_string(&holder(2u32.pow(22))).unwrap()).unwrap();
        let options = LockOptions {
            stale_after_ms: Some(1),
            ..Default::default()
        };

        let stale = DirLock::inspect(&stale_path, options).unwrap();
        let dead = DirLock::inspect(&dead_path, options).unwrap();

        assert!(matches!(stale, LockState::LeaseExpired(_)));
        assert!(matches!(dead, LockState::HolderGone(_)));
        assert!(stale.blocks_a_writer(), "the hand-over grace is still owed");
        assert!(!dead.blocks_a_writer(), "nothing is left there to give notice to");
    }

    #[test]
    fn inspect_reads_a_lock_file_it_cannot_parse_as_unreadable_and_refuses_a_planted_link() {
        let dir = temp_dir("hayson-lock-inspect-unreadable-");
        let path = dir.join(LOCK_FILE);
        let linked = dir.join("linked.lock");
        fs::write(&path, "{ not a holder }").unwrap();
        symlink(dir.join("elsewhere"), &linked).unwrap();

        let unreadable = DirLock::inspect(&path, LockOptions::default()).unwrap();
        let planted = DirLock::inspect(&linked, LockOptions::default());

        assert_eq!(unreadable, LockState::Unreadable);
        assert!(unreadable.blocks_a_writer(), "it is reclaimed only once it goes stale");
        assert!(matches!(planted, Err(DirLockError::SuspiciousLockFile(_))));
    }

    #[test]
    fn never_steals_from_a_holder_that_keeps_beating_across_several_lease_periods() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-no-false-steal-");
        let path = dir.join(LOCK_FILE);
        let mut held = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(120),
                heartbeat_ms: Some(20),
                ..Default::default()
            },
        )
        .unwrap();
        let owner = held.owner().to_string();

        for _ in 0..4 {
            let refused = DirLock::acquire(
                path.clone(),
                LockOptions {
                    stale_after_ms: Some(120),
                    timeout_ms: Some(150),
                    retry_interval_ms: Some(10),
                    ..Default::default()
                },
            );
            assert!(matches!(refused, Err(DirLockError::Locked(_))));
            assert_eq!(holder_of(&path).owner, owner);
        }

        assert!(held.tick().is_ok());
        assert!(!held.is_lost());
        held.release();
    }

    /// A holder that goes quiet for longer than a lease but comes back inside the grace window
    /// keeps its lock: the challenger sees the heartbeat move and calls the steal off.
    #[test]
    fn a_holder_that_resumes_beating_inside_the_grace_window_keeps_its_lock() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-grace-recovery-");
        let path = dir.join(LOCK_FILE);
        let owner = mint_owner();
        let counter = AtomicU64::new(0);
        let stale = LockHolder {
            owner: owner.clone(),
            pid: Some(std::process::id()),
            host: platform::hostname(),
            started_at: 0,
            heartbeat_at: 0,
        };
        fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();

        let beating_path = path.clone();
        let beating_owner = owner.clone();
        let resumed = thread::spawn(move || {
            thread::sleep(Duration::from_millis(150));
            let beaten = LockHolder {
                heartbeat_at: now_ms(),
                ..stale
            };
            let temp = write_temp(&beating_path, &beating_owner, &counter, &beaten).unwrap();
            fs::rename(&temp, &beating_path).unwrap();
        });

        let refused = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(600),
                timeout_ms: Some(400),
                retry_interval_ms: Some(10),
                ..Default::default()
            },
        );
        resumed.join().unwrap();
        assert!(matches!(refused, Err(DirLockError::Locked(_))));
        assert_eq!(holder_of(&path).owner, owner);
    }

    #[test]
    fn a_dispossessed_holder_learns_it_lost_the_lock_on_its_next_tick() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-fencing-");
        let path = dir.join(LOCK_FILE);
        let mut robbed = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(120),
                // NOTE: a heartbeat far longer than the test keeps the thread out of the way, so
                // the steal below is not racing a renewal it never sees.
                heartbeat_ms: Some(60_000),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(robbed.tick().is_ok());

        thread::sleep(Duration::from_millis(200));
        let mut thief = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(120),
                timeout_ms: Some(2_000),
                retry_interval_ms: Some(10),
                ..Default::default()
            },
        )
        .unwrap();
        assert_ne!(thief.owner(), robbed.owner());

        // NOTE: forcing the beat due is what a real handle gets for free after heartbeat_ms; this
        // test wants the fence itself, not the wait, so it does not sleep out a real interval.
        robbed.beat.last_beat_ms.store(0, Ordering::SeqCst);
        let fenced = robbed.tick();
        assert!(
            matches!(fenced, Err(DirLockError::Lost(_))),
            "expected Lost, got {fenced:?}"
        );
        assert!(robbed.is_lost());
        assert!(matches!(robbed.tick(), Err(DirLockError::Lost(_))));

        // A lost handle must not take the thief's lock file away with it.
        robbed.release();
        assert_eq!(holder_of(&path).owner, thief.owner());
        thief.release();
    }

    #[test]
    fn exactly_one_of_many_concurrent_challengers_takes_an_expired_lock() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-steal-race-");
        let path = dir.join(LOCK_FILE);
        let stale = LockHolder {
            owner: mint_owner(),
            pid: Some(std::process::id()),
            host: platform::hostname(),
            started_at: 0,
            heartbeat_at: 0,
        };
        fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();

        let mut challengers = Vec::new();
        for _ in 0..8 {
            let path = path.clone();
            challengers.push(thread::spawn(move || {
                DirLock::acquire(
                    path,
                    LockOptions {
                        stale_after_ms: Some(200),
                        heartbeat_ms: Some(60_000),
                        timeout_ms: Some(400),
                        retry_interval_ms: Some(5),
                        grace_served: None,
                    },
                )
            }));
        }

        let mut winners = Vec::new();
        for challenger in challengers {
            match challenger.join().unwrap() {
                Ok(lock) => winners.push(lock),
                Err(DirLockError::Locked(_)) => {}
                Err(other) => panic!("unexpected failure: {other}"),
            }
        }

        // The rename cannot be made a compare-and-swap, so a challenger whose write was superseded
        // in the moment between its own rename and its read-back can still come back believing it
        // won. The invariant is one step further on: the lock file names exactly one of them, and
        // every other one is fenced the first time it asks.
        assert!(!winners.is_empty(), "an expired lock was left to nobody");
        let landed = holder_of(&path).owner;
        let mut holding = 0;
        for winner in &mut winners {
            winner.beat.last_beat_ms.store(0, Ordering::SeqCst);
            if winner.owner() == landed {
                holding += 1;
                assert!(winner.tick().is_ok(), "the writer named in the lock file was fenced");
            } else {
                assert!(
                    matches!(winner.tick(), Err(DirLockError::Lost(_))),
                    "a challenger that did not land in the lock file was not fenced",
                );
            }
        }
        assert_eq!(holding, 1, "more than one challenger owns the same lock file");

        for winner in &mut winners {
            winner.release();
        }
        assert!(!path.exists(), "the surviving writer did not clean up after itself");
    }

    /// Writes an expired lease nobody will renew, the way a writer killed mid-flight leaves one.
    fn abandoned_lease(path: &Path) -> LockHolder {
        let stale = LockHolder {
            owner: mint_owner(),
            // NOTE: this process's own live pid — a pid check would call it alive and never steal.
            pid: Some(std::process::id()),
            host: platform::hostname(),
            started_at: 0,
            heartbeat_at: 0,
        };
        fs::write(path, serde_json::to_string(&stale).unwrap()).unwrap();
        stale
    }

    fn grace_owed(outcome: ChallengeOutcome) -> (LockHolder, u64) {
        match outcome {
            ChallengeOutcome::GraceOwed { holder, grace_ms } => (holder, grace_ms),
            other => panic!("expected GraceOwed, got {other:?}"),
        }
    }

    /// The whole point of the split: the pass that finds an expired lease comes straight back with
    /// the grace it owes, instead of sleeping it out inside the call.
    #[test]
    fn challenge_reports_the_grace_an_expired_lease_is_owed_without_waiting_any_of_it() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-challenge-");
        let path = dir.join(LOCK_FILE);
        let stale = abandoned_lease(&path);
        let options = LockOptions {
            stale_after_ms: Some(4_000),
            ..Default::default()
        };

        let started_at = Instant::now();
        let (holder, grace_ms) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());
        let answered_in = started_at.elapsed();

        assert_eq!(holder.owner, stale.owner);
        assert_eq!(grace_ms, 2_000);
        assert!(
            answered_in < Duration::from_millis(500),
            "the challenge waited {answered_in:?} of the grace it was supposed to hand back",
        );

        // The caller owns the wait, and the reclaim lands on the far side of it.
        thread::sleep(Duration::from_millis(grace_ms.min(120)));
        let mut lock = DirLock::finish_challenge(path.clone(), options, &holder)
            .unwrap()
            .unwrap();
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }

    /// A free directory and a provably dead holder owe nobody notice, so a challenge answers with
    /// the lock itself and there is no second call to make.
    #[test]
    fn challenge_takes_a_free_lock_outright_and_reports_a_live_holder_as_blocked() {
        let dir = temp_dir("hayson-lock-challenge-free-");
        let path = dir.join(LOCK_FILE);
        let options = LockOptions::default();

        let taken = DirLock::challenge(path.clone(), options).unwrap();
        let mut first = match taken {
            ChallengeOutcome::Taken(lock) => lock,
            other => panic!("expected Taken, got {other:?}"),
        };

        match DirLock::challenge(path.clone(), options).unwrap() {
            ChallengeOutcome::Blocked(Some(holder)) => assert_eq!(holder.owner, first.owner()),
            other => panic!("expected Blocked, got {other:?}"),
        }
        first.release();
    }

    /// The safety property the split has to keep: two challengers can hold the same expired lease
    /// in hand, wait the same grace out, and still only one of them ends up with the lock file.
    #[test]
    fn of_two_callers_finishing_one_challenge_exactly_one_takes_the_lock_and_the_other_is_refused() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-finish-race-");
        let path = dir.join(LOCK_FILE);
        abandoned_lease(&path);
        let options = LockOptions {
            stale_after_ms: Some(200),
            heartbeat_ms: Some(60_000),
            ..Default::default()
        };

        let (first_seen, _) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());
        let (second_seen, _) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());
        assert_eq!(first_seen, second_seen, "both challengers read the same lease");

        thread::sleep(Duration::from_millis(120));
        let winner = DirLock::finish_challenge(path.clone(), options, &first_seen).unwrap();
        let loser = DirLock::finish_challenge(path.clone(), options, &second_seen).unwrap();

        let mut winner = winner.expect("the first caller to finish the challenge got nothing");
        assert!(
            loser.is_none(),
            "the second caller took a lock the first had already reclaimed",
        );
        assert_eq!(holder_of(&path).owner, winner.owner());
        winner.release();
        assert!(!path.exists(), "the surviving writer did not clean up after itself");
    }

    /// The same race run for real, from threads that reach the rename together. The rename cannot
    /// be made a compare-and-swap, so the invariant is the one `acquire` already rests on: the
    /// lock file names exactly one of them, and every other is fenced the first time it asks.
    #[test]
    fn concurrent_finishers_of_one_challenge_leave_the_lock_file_naming_exactly_one_of_them() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-finish-threads-");
        let path = dir.join(LOCK_FILE);
        abandoned_lease(&path);
        let options = LockOptions {
            stale_after_ms: Some(200),
            heartbeat_ms: Some(60_000),
            ..Default::default()
        };

        let mut finishers = Vec::new();
        for _ in 0..8 {
            let path = path.clone();
            finishers.push(thread::spawn(move || {
                let (holder, _) = match DirLock::challenge(path.clone(), options).unwrap() {
                    ChallengeOutcome::GraceOwed { holder, grace_ms } => (holder, grace_ms),
                    other => panic!("expected GraceOwed, got {other:?}"),
                };
                thread::sleep(Duration::from_millis(120));
                DirLock::finish_challenge(path, options, &holder).unwrap()
            }));
        }

        let mut winners: Vec<DirLock> = finishers
            .into_iter()
            .filter_map(|finisher| finisher.join().unwrap())
            .collect();

        assert!(!winners.is_empty(), "an expired lock was left to nobody");
        let landed = holder_of(&path).owner;
        let mut holding = 0;
        for winner in &mut winners {
            winner.beat.last_beat_ms.store(0, Ordering::SeqCst);
            if winner.owner() == landed {
                holding += 1;
                assert!(winner.tick().is_ok(), "the writer named in the lock file was fenced");
            } else {
                assert!(
                    matches!(winner.tick(), Err(DirLockError::Lost(_))),
                    "a finisher that did not land in the lock file was not fenced",
                );
            }
        }
        assert_eq!(holding, 1, "more than one finisher owns the same lock file");

        for winner in &mut winners {
            winner.release();
        }
        assert!(!path.exists(), "the surviving writer did not clean up after itself");
    }

    /// A holder that comes back inside the caller's own grace window keeps its lock, exactly as it
    /// does when `acquire` sleeps the grace itself: the token names a lease that has moved on.
    #[test]
    fn finishing_a_challenge_a_resumed_holder_has_moved_past_reclaims_nothing() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-finish-resumed-");
        let path = dir.join(LOCK_FILE);
        let stale = abandoned_lease(&path);
        let options = LockOptions {
            stale_after_ms: Some(600),
            ..Default::default()
        };

        let (holder, _) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());

        let counter = AtomicU64::new(0);
        let beaten = LockHolder {
            heartbeat_at: now_ms(),
            ..stale.clone()
        };
        let temp = write_temp(&path, &stale.owner, &counter, &beaten).unwrap();
        fs::rename(&temp, &path).unwrap();

        assert!(
            DirLock::finish_challenge(path.clone(), options, &holder)
                .unwrap()
                .is_none(),
            "a holder that proved itself alive inside the grace lost its lock anyway",
        );
        assert_eq!(holder_of(&path).owner, stale.owner);
    }

    /// `LockOptions::grace_served` is how a host that can only reach `acquire` gets at
    /// `finish_challenge`: a fail-fast open that would report the directory locked reclaims the
    /// very lease the token names, and sleeps none of the grace doing it.
    #[test]
    fn a_fail_fast_acquire_carrying_a_served_grace_reclaims_the_lease_that_token_names() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-served-");
        let path = dir.join(LOCK_FILE);
        abandoned_lease(&path);
        let looking = LockOptions {
            stale_after_ms: Some(4_000),
            timeout_ms: Some(0),
            ..Default::default()
        };

        // Without the token there is no room in a fail-fast budget for a hand-over at all.
        let refused = DirLock::acquire(path.clone(), looking).unwrap_err();
        assert!(matches!(refused, DirLockError::Locked(_)), "got {refused:?}");

        let (holder, grace_ms) = grace_owed(DirLock::challenge(path.clone(), looking).unwrap());
        assert_eq!(grace_ms, 2_000);
        thread::sleep(Duration::from_millis(120));

        let started_at = Instant::now();
        let mut lock = DirLock::acquire(
            path.clone(),
            LockOptions {
                grace_served: GraceServed::of(&holder),
                ..looking
            },
        )
        .unwrap();
        assert!(
            started_at.elapsed() < Duration::from_millis(500),
            "the open slept a grace its caller had already served",
        );
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }

    /// The token is spent on the lease it names and nothing else. One naming a lease that has
    /// since changed hands buys nothing, and a fail-fast open says locked rather than reclaiming.
    #[test]
    fn a_served_grace_naming_another_lease_reclaims_nothing() {
        let _no_pids = without_pid_liveness();
        let dir = temp_dir("hayson-lock-served-stale-token-");
        let path = dir.join(LOCK_FILE);
        let first = abandoned_lease(&path);
        let served = GraceServed::of(&first);
        assert!(served.is_some());

        // A second writer took over meanwhile, and left an expired lease of its own behind.
        let second = abandoned_lease(&path);
        assert_ne!(second.owner, first.owner);

        let refused = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(4_000),
                timeout_ms: Some(0),
                grace_served: served,
                ..Default::default()
            },
        )
        .unwrap_err();
        assert!(matches!(refused, DirLockError::Locked(_)), "got {refused:?}");
        assert_eq!(holder_of(&path).owner, second.owner);
    }

    #[test]
    fn a_served_grace_is_only_read_off_an_owner_token_this_engine_minted() {
        let minted = mint_owner();
        assert!(GraceServed::parse(&minted, 7).is_some());
        assert_eq!(GraceServed::parse(&minted, 7), GraceServed::parse(&minted, 7));
        assert_ne!(GraceServed::parse(&minted, 7), GraceServed::parse(&minted, 8));
        assert!(GraceServed::parse("", 7).is_none());
        assert!(GraceServed::parse("not-a-token", 7).is_none());
        assert!(GraceServed::parse(&minted[..31], 7).is_none());
        assert!(GraceServed::parse("zz000000000000000000000000000000", 7).is_none());
    }

    #[test]
    fn reclaims_an_unparsable_lock_file_nobody_touched_for_longer_than_stale_after_ms() {
        let dir = temp_dir("hayson-lock-unparsable-stale-");
        let path = dir.join(LOCK_FILE);
        fs::write(&path, "{\"pid\":48213,\"host\":\"thi").unwrap();
        let past = SystemTime::now() - Duration::from_secs(60);
        let file = fs::File::options().write(true).open(&path).unwrap();
        file.set_times(fs::FileTimes::new().set_modified(past)).unwrap();

        let mut lock = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(1_000),
                timeout_ms: Some(200),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }

    /// A lock file in the pre-owner format has no token to compare, so it is unparsable by
    /// construction and takes the reclaim-after-`stale_after_ms` path. That is the whole migration.
    #[test]
    fn an_old_format_pid_only_lock_file_is_unparsable_and_reclaimed_after_the_lease() {
        let dir = temp_dir("hayson-lock-old-format-");
        let path = dir.join(LOCK_FILE);
        let old = serde_json::json!({
            "pid": std::process::id(),
            "host": platform::hostname(),
            "started_at": now_ms(),
            "heartbeat_at": now_ms(),
        });
        fs::write(&path, serde_json::to_string(&old).unwrap()).unwrap();
        assert!(serde_json::from_str::<LockHolder>(&fs::read_to_string(&path).unwrap()).is_err());

        let fresh = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(30_000),
                timeout_ms: Some(100),
                retry_interval_ms: Some(10),
                ..Default::default()
            },
        );
        assert!(
            matches!(fresh, Err(DirLockError::Locked(_))),
            "a fresh unparsable file is left alone"
        );

        let past = SystemTime::now() - Duration::from_secs(60);
        let file = fs::File::options().write(true).open(&path).unwrap();
        file.set_times(fs::FileTimes::new().set_modified(past)).unwrap();

        let mut lock = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(1_000),
                timeout_ms: Some(200),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }

    #[test]
    fn leaves_a_freshly_written_unparsable_lock_file_alone() {
        let dir = temp_dir("hayson-lock-unparsable-fresh-");
        let path = dir.join(LOCK_FILE);
        fs::write(&path, "").unwrap();

        let failure = DirLock::acquire(
            path,
            LockOptions {
                stale_after_ms: Some(30_000),
                timeout_ms: Some(100),
                retry_interval_ms: Some(10),
                ..Default::default()
            },
        )
        .unwrap_err();
        assert!(matches!(failure, DirLockError::Locked(_)));
    }

    #[test]
    fn refuses_to_touch_a_lock_path_somebody_replaced_with_a_symlink() {
        let dir = temp_dir("hayson-lock-symlink-");
        let victim_dir = temp_dir("hayson-lock-victim-");
        let victim = victim_dir.join("precious.txt");
        fs::write(&victim, "PRECIOUS\n").unwrap();
        let victim_mtime = fs::metadata(&victim).unwrap().modified().unwrap();

        let path = dir.join(LOCK_FILE);
        symlink(&victim, &path).unwrap();

        let failure = DirLock::acquire(
            path.clone(),
            LockOptions {
                stale_after_ms: Some(0),
                timeout_ms: Some(100),
                ..Default::default()
            },
        )
        .unwrap_err();
        assert!(
            matches!(failure, DirLockError::SuspiciousLockFile(ref at) if at == &path),
            "expected SuspiciousLockFile, got {failure:?}",
        );

        assert_eq!(fs::read_to_string(&victim).unwrap(), "PRECIOUS\n");
        assert_eq!(fs::metadata(&victim).unwrap().modified().unwrap(), victim_mtime);
        assert!(fs::symlink_metadata(&path).unwrap().file_type().is_symlink());
    }

    /// The heartbeat's O_EXCL-and-rename design, with no `O_NOFOLLOW` anywhere near it: every
    /// candidate temporary an attacker has already claimed is refused and skipped, the file behind
    /// the planted link is never touched, and the beat lands on the first free name.
    #[test]
    fn the_heartbeat_refuses_every_planted_temporary_and_beats_on_a_free_name() {
        let dir = temp_dir("hayson-heartbeat-symlink-");
        let path = dir.join(LOCK_FILE);
        let mut lock = DirLock::acquire(
            path.clone(),
            LockOptions {
                heartbeat_ms: Some(60_000),
                ..Default::default()
            },
        )
        .unwrap();

        let victim_dir = temp_dir("hayson-heartbeat-victim-");
        let victim = victim_dir.join("precious.txt");
        fs::write(&victim, "PRECIOUS\n").unwrap();
        let victim_mtime = fs::metadata(&victim).unwrap().modified().unwrap();

        let planted: Vec<PathBuf> = (0..3).map(|n| lock.temp_path_at(n)).collect();
        for temp in &planted {
            symlink(&victim, temp).unwrap();
        }

        let before = holder_of(&path).heartbeat_at;
        thread::sleep(Duration::from_millis(2));
        lock.beat.renew().unwrap();

        assert_eq!(fs::read_to_string(&victim).unwrap(), "PRECIOUS\n");
        assert_eq!(fs::metadata(&victim).unwrap().modified().unwrap(), victim_mtime);
        for temp in &planted {
            assert!(
                fs::symlink_metadata(temp).unwrap().file_type().is_symlink(),
                "a planted link was consumed instead of skipped",
            );
        }
        assert!(holder_of(&path).heartbeat_at > before);
        assert_eq!(holder_of(&path).owner, lock.owner());
        assert!(!fs::symlink_metadata(&path).unwrap().file_type().is_symlink());
        assert!(
            !lock.temp_path_at(3).exists(),
            "the temporary must not outlive the rename"
        );

        lock.release();
        for temp in &planted {
            let _ = fs::remove_file(temp);
        }
    }

    #[test]
    fn the_background_heartbeat_keeps_the_lease_fresh_on_its_own() {
        let dir = temp_dir("hayson-heartbeat-runs-");
        let path = dir.join(LOCK_FILE);
        let mut lock = DirLock::acquire(
            path.clone(),
            LockOptions {
                heartbeat_ms: Some(20),
                ..Default::default()
            },
        )
        .unwrap();

        let before = holder_of(&path).heartbeat_at;
        let beaten = until(
            || {
                let seen = holder_of(&path).heartbeat_at;
                (seen > before).then_some(seen)
            },
            5_000,
        );
        assert!(beaten.is_some(), "the background heartbeat never advanced the lease");
        assert_eq!(holder_of(&path).owner, lock.owner());
        lock.release();
    }
}