newt-core 0.8.0

//! Crash-safe, process-serialized file writes.
//!
//! The lock owner is recorded in the sidecar so a process killed without
//! running `Drop` does not wedge the destination forever.  Writes are staged
//! beside the destination, synced, atomically replaced, and followed by a
//! parent-directory sync so a successful return is durable across a crash.

// Model: GPT-5 | Harness: Codex | Operator: Shawn Hartsock | Time: 18:32 EDT | Date: 2026-08-12

use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const LOCK_RETRIES: u32 = 100;
const LOCK_RETRY_DELAY: Duration = Duration::from_millis(20);
const LEGACY_LOCK_STALE: Duration = Duration::from_secs(30);
/// How long an empty lock must sit before it is reclaimed. This version
/// never observes its own lock empty (see [`publish_lock`]), so an empty
/// lock on disk can only be a crash artifact of an OLDER build's
/// `create_new` + `write_all` sequence — but that older build's own writer
/// is also empty for the instant between those two calls, so the grace must
/// outlast any real write window (microseconds) while staying well inside
/// the `LOCK_RETRIES` * `LOCK_RETRY_DELAY` contention budget (~2s), so a
/// contender still gets a live/dead answer within that budget instead of
/// exhausting it waiting out the grace.
const EMPTY_LOCK_GRACE: Duration = Duration::from_millis(200);
static UNIQUE_COUNTER: AtomicU64 = AtomicU64::new(0);

/// A durable replacement failure that records whether the destination name
/// was already changed. Callers which publish multi-file invariants must not
/// roll back prerequisites after [`Self::committed`] becomes true.
#[derive(Debug)]
pub struct DurableReplaceError {
    path: PathBuf,
    committed: bool,
    source: std::io::Error,
}

impl DurableReplaceError {
    fn before_commit(path: &Path, source: std::io::Error) -> Self {
        Self {
            path: path.to_path_buf(),
            committed: false,
            source,
        }
    }

    /// Build an error for a failure after the destination name changed.
    ///
    /// This is public so transaction adapters and deterministic failpoints can
    /// preserve the same commit-state contract as [`ResolvedPath`].
    pub fn after_commit(path: &Path, source: std::io::Error) -> Self {
        Self {
            path: path.to_path_buf(),
            committed: true,
            source,
        }
    }

    /// Whether replacement changed the destination name before failing.
    #[must_use]
    pub fn committed(&self) -> bool {
        self.committed
    }
}

impl std::fmt::Display for DurableReplaceError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if self.committed {
            write!(
                formatter,
                "replaced {}, but could not durably sync its parent directory: {}",
                self.path.display(),
                self.source
            )
        } else {
            write!(
                formatter,
                "could not replace {}: {}",
                self.path.display(),
                self.source
            )
        }
    }
}

impl std::error::Error for DurableReplaceError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(&self.source)
    }
}

/// Kernel-held serialization for stale-lock takeover. The file is deliberately
/// persistent: deleting a native-lock inode while another contender has it
/// open would recreate the same generation race this lease prevents. Closing
/// the file releases the OS lock automatically, including after process death.
struct ReclaimLease {
    _file: std::fs::File,
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct LockOwner {
    pid: i64,
    nonce: String,
}

impl LockOwner {
    fn current() -> Self {
        Self {
            pid: i64::from(std::process::id()),
            nonce: unique_suffix(),
        }
    }

    fn encode(&self) -> String {
        format!("{}:{}\n", self.pid, self.nonce)
    }

    fn decode(body: &str) -> Option<Self> {
        let (pid, nonce) = body.trim().split_once(':')?;
        let pid: i64 = pid.parse().ok()?;
        if pid <= 0 {
            return None;
        }
        if nonce.is_empty() {
            return None;
        }
        Some(Self {
            pid,
            nonce: nonce.to_string(),
        })
    }
}

/// Held for the lifetime of a locked critical section.
///
/// Drop removes the sidecar only when it still contains this guard's owner
/// token.  A late drop can therefore never unlink a replacement owner's lock.
#[must_use = "drop the guard to release the lock"]
#[derive(Debug)]
pub struct LockGuard {
    path: PathBuf,
    owner: LockOwner,
}

impl Drop for LockGuard {
    fn drop(&mut self) {
        let ours = std::fs::read_to_string(&self.path)
            .ok()
            .and_then(|body| LockOwner::decode(&body))
            .is_some_and(|owner| owner == self.owner);
        if ours {
            let _ = std::fs::remove_file(&self.path);
            let _ = sync_parent(&self.path);
        }
    }
}

/// Acquire the advisory lock at `lock_path`.
///
/// A parseable lock owned by a dead process is reclaimed immediately.  Older
/// lock files written by Newt versions which did not record an owner retain the
/// conservative age-based recovery path.
pub fn acquire_lock(lock_path: &Path) -> anyhow::Result<LockGuard> {
    if let Some(parent) = lock_path.parent() {
        if !parent.as_os_str().is_empty() {
            std::fs::create_dir_all(parent)?;
        }
    }
    sweep_lock_tmp_orphans(lock_path);
    let owner = LockOwner::current();
    let encoded = owner.encode();
    let mut denied = false;
    for _ in 0..LOCK_RETRIES {
        match publish_lock(lock_path, &encoded) {
            Ok(()) => {
                return Ok(LockGuard {
                    path: lock_path.to_path_buf(),
                    owner,
                });
            }
            // Windows can report a just-deleted lock as DELETE_PENDING via
            // ERROR_ACCESS_DENIED.  Retry it like ordinary contention.
            Err(error)
                if matches!(
                    error.kind(),
                    std::io::ErrorKind::AlreadyExists | std::io::ErrorKind::PermissionDenied
                ) =>
            {
                denied = error.kind() == std::io::ErrorKind::PermissionDenied;
                if reclaim_lock_once(lock_path)? {
                    continue;
                }
                std::thread::sleep(LOCK_RETRY_DELAY);
            }
            Err(error) => return Err(error.into()),
        }
    }
    if denied {
        anyhow::bail!(
            "could not acquire {} — permission denied; check the lock directory permissions",
            lock_path.display()
        );
    }
    match current_owner(lock_path) {
        Some(owner) => anyhow::bail!(
            "could not acquire {} — {CONTENDED} (held by pid {})",
            lock_path.display(),
            owner.pid
        ),
        None => anyhow::bail!("could not acquire {} — {CONTENDED}", lock_path.display()),
    }
}

/// Remove `<lock>.*.tmp` orphans left by [`publish_lock`] crashing between
/// its temp-file write and the `hard_link`/fallback publish. Best-effort and
/// scoped to files older than the contention retry budget, so an in-flight
/// writer's own temp file is never touched; errors are swallowed since a
/// stray orphan blocks nothing and this sweep must never block acquisition.
fn sweep_lock_tmp_orphans(lock_path: &Path) {
    sweep_lock_tmp_orphans_older_than(lock_path, LOCK_RETRY_DELAY * LOCK_RETRIES);
}

/// [`sweep_lock_tmp_orphans`] with the age budget as a seam, so a test can
/// use a millisecond-scale budget instead of the real ~2s retry budget.
fn sweep_lock_tmp_orphans_older_than(lock_path: &Path, budget: Duration) {
    let Some(parent) = lock_path.parent().filter(|p| !p.as_os_str().is_empty()) else {
        return;
    };
    let Some(file_name) = lock_path
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
    else {
        return;
    };
    let prefix = format!("{file_name}.");
    let Ok(entries) = std::fs::read_dir(parent) else {
        return;
    };
    for entry in entries.flatten() {
        let name = entry.file_name();
        let name = name.to_string_lossy();
        if !name.starts_with(&prefix) || !name.ends_with(".tmp") {
            continue;
        }
        let is_old = entry
            .metadata()
            .and_then(|metadata| metadata.modified())
            .ok()
            .and_then(|modified| modified.elapsed().ok())
            .is_some_and(|age| age > budget);
        if is_old {
            let _ = std::fs::remove_file(entry.path());
        }
    }
}

/// Publish `encoded` at `lock_path` so it is never observable empty.
///
/// Writes the owner body to a private temp file first, `fsync`s it, then
/// atomically publishes it with `hard_link` — which fails with
/// `AlreadyExists` exactly when `create_new` would have, so callers get the
/// same contention semantics. Unlike `create_new` + `write_all`, there is no
/// window between "the lock file exists" and "the lock file has content": a
/// lock this function writes carries its final body from the instant it
/// becomes visible under `lock_path`. `hard_link` needs the source and
/// destination on the same filesystem, which they are (`lock_path`'s own
/// directory); if a future target lacks hard-link support, fall back to the
/// old `create_new` + `write_all` path instead of failing outright.
fn publish_lock(lock_path: &Path, encoded: &str) -> std::io::Result<()> {
    publish_lock_with_link(lock_path, encoded, |src, dst| std::fs::hard_link(src, dst))
}

/// [`publish_lock`] with the `hard_link` call as a seam, so a test can inject
/// a link failure without a real vfat/FUSE/SMB mount.
fn publish_lock_with_link(
    lock_path: &Path,
    encoded: &str,
    link: impl FnOnce(&Path, &Path) -> std::io::Result<()>,
) -> std::io::Result<()> {
    let dir = lock_path.parent().filter(|p| !p.as_os_str().is_empty());
    let tmp_name = lock_path
        .file_name()
        .map(|name| format!("{}.{}.tmp", name.to_string_lossy(), unique_suffix()))
        .unwrap_or_else(|| format!("lock.{}.tmp", unique_suffix()));
    let tmp_path = match dir {
        Some(dir) => dir.join(tmp_name),
        None => PathBuf::from(tmp_name),
    };
    let write_result = (|| -> std::io::Result<()> {
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&tmp_path)?;
        file.write_all(encoded.as_bytes())?;
        file.sync_all()
    })();
    if let Err(error) = write_result {
        let _ = std::fs::remove_file(&tmp_path);
        return Err(error);
    }
    let link_result = link(&tmp_path, lock_path);
    match &link_result {
        Ok(()) => {
            let _ = std::fs::remove_file(&tmp_path);
        }
        Err(error) if error.kind() != std::io::ErrorKind::AlreadyExists => {
            // hard_link failed for any reason other than "the lock is
            // already held" — not just Unsupported. vfat/exFAT/FUSE(sshfs)
            // return EPERM (which io::ErrorKind maps to PermissionDenied,
            // the same kind ordinary contention/Windows DELETE_PENDING
            // reports) and some Windows FAT/SMB errors map to Uncategorized.
            // Treating only Unsupported as "no hard-link support" left those
            // targets either misdiagnosed as contention or failing outright.
            // Fall back to the old direct create_new path, same contention
            // semantics, just without the empty-window guarantee; the
            // fallback's own error (if any) is the verdict callers see.
            let _ = std::fs::remove_file(&tmp_path);
            return std::fs::OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(lock_path)
                .and_then(|mut file| {
                    file.write_all(encoded.as_bytes())?;
                    file.sync_all()
                });
        }
        Err(_) => {
            let _ = std::fs::remove_file(&tmp_path);
        }
    }
    link_result
}

/// The owner recorded in `lock_path` right now, for the refusal message.
/// Best-effort: a lock that vanished or was rewritten between the last
/// failed retry and this read reports `None` rather than lying about who
/// holds it.
fn current_owner(lock_path: &Path) -> Option<LockOwner> {
    LockOwner::decode(&std::fs::read_to_string(lock_path).ok()?)
}

/// The refusal [`acquire_lock`] returns when another LIVE owner holds the lock
/// and the retry budget ran out.
///
/// One constant, so the message a human reads and the predicate a caller
/// branches on cannot drift apart.
const CONTENDED: &str = "another live process appears to be writing it; try again";

/// Whether `error` is [`acquire_lock`] reporting CONTENTION — the one failure
/// it invites the caller to retry.
///
/// Until #1871 "try again" was advice in PROSE. A caller that wanted to act on
/// it had to match the message itself, so none did, and the test that most
/// needed to — six concurrent approvals racing one lock — `unwrap()`ed instead.
/// That made a correctly serialized refusal indistinguishable from a lost
/// update, which is the whole of that flake: the registry was right and the
/// assertion was wrong.
///
/// Deliberately NOT true for the permission-denied refusal. That one is a
/// misconfigured lock directory; retrying it just fails slower.
#[must_use]
pub fn is_lock_contended(error: &anyhow::Error) -> bool {
    error
        .chain()
        .any(|cause| cause.to_string().contains(CONTENDED))
}

#[cfg(test)]
thread_local! {
    /// How many times THIS thread's in-flight `acquire_lock` call has invoked
    /// [`reclaim_lock_once`] — the count-based alternative to timing the call
    /// (`killed_process_lock_is_reclaimed` asserts this instead of elapsed
    /// time). Thread-local, not a global `static`: tests on other threads
    /// never pollute it, no serialization needed beyond what `real_fs`
    /// already requires for the filesystem itself.
    static RECLAIM_ATTEMPTS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

/// Reset [`RECLAIM_ATTEMPTS`] before a call under test.
#[cfg(test)]
fn reset_reclaim_attempts_for_test() {
    RECLAIM_ATTEMPTS.with(|count| count.set(0));
}

/// How many reclaim attempts the most recent call made on this thread.
#[cfg(test)]
fn reclaim_attempts_for_test() -> usize {
    RECLAIM_ATTEMPTS.with(std::cell::Cell::get)
}

#[cfg(test)]
thread_local! {
    /// Overrides [`reclaimable_lock`]'s `now` for the duration of a test.
    /// Thread-local, same reasoning as [`RECLAIM_ATTEMPTS`]: no cross-test
    /// serialization needed beyond what `real_fs` already requires.
    ///
    /// Why this exists: counting attempts (above) proves the owner-aware,
    /// live-PID branch fired instead of the slow legacy age-based fallback —
    /// but only if `age()` reads a value close to when the lock was written.
    /// Without pinning, a deliberately broken implementation that folds the
    /// age check into the OWNER branch could still pass by coincidence: if
    /// the real wall clock has, by the time `acquire_lock` finally calls
    /// `reclaim_lock_once`, already crossed `LEGACY_LOCK_STALE` past the
    /// lock's real write time (an unlikely but unbounded delay — process
    /// spawn/kill/wait, scheduler contention), the broken branch's age check
    /// would ALSO be satisfied on attempt 1, matching the correct
    /// implementation's attempt count for the wrong reason. Pinning `now` to
    /// the lock file's modification timestamp makes `age()` exactly zero
    /// regardless of how long the surrounding test machinery actually takes
    /// in real time, closing that gap.
    static PINNED_NOW: std::cell::Cell<Option<std::time::SystemTime>> =
        const { std::cell::Cell::new(None) };
}

/// Pin [`reclaimable_lock`]'s `now` to `now` for the returned guard's
/// lifetime; real [`std::time::SystemTime::now`] resumes on drop (including
/// on an early return or a panic, so a failing assertion cannot leak the pin
/// into a later test on this thread).
#[cfg(test)]
#[must_use]
fn pin_now_for_test(now: std::time::SystemTime) -> impl Drop {
    struct Unpin;
    impl Drop for Unpin {
        fn drop(&mut self) {
            PINNED_NOW.with(|cell| cell.set(None));
        }
    }
    PINNED_NOW.with(|cell| cell.set(Some(now)));
    Unpin
}

/// Attempt one stale-generation takeover while holding a kernel advisory lock
/// on a persistent recovery sidecar. Every reclaimer must hold this lease
/// across both the liveness proof and unlink, so a second reclaimer cannot act
/// on an observation from the previous generation.
fn reclaim_lock_once(lock_path: &Path) -> anyhow::Result<bool> {
    #[cfg(test)]
    RECLAIM_ATTEMPTS.with(|count| count.set(count.get() + 1));
    let Some(_lease) = try_reclaim_lease(lock_path)? else {
        return Ok(false);
    };
    #[cfg(test)]
    pause_reclaimer_for_test()?;
    if !reclaimable_lock(lock_path) {
        return Ok(false);
    }
    // Pre-existing race, not introduced here: this removes by PATH, so if the
    // dead owner's guard already unlinked it and a third process published a
    // live lock in the gap between the read above and this unlink, we delete
    // the LIVE lock instead. The lease only serializes reclaimers against each
    // other, not against owner-drop + publish. Narrowing this needs a re-read
    // (or dev/ino compare) under the lease immediately before unlinking —
    // tracked as a follow-up, not fixed here.
    match std::fs::remove_file(lock_path) {
        Ok(()) => {
            sync_parent(lock_path)?;
            Ok(true)
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
        Err(error) => Err(error.into()),
    }
}

#[cfg(test)]
fn pause_reclaimer_for_test() -> anyhow::Result<()> {
    const READY_ENV: &str = "NEWT_ATOMIC_RECLAIM_READY";
    const RELEASE_ENV: &str = "NEWT_ATOMIC_RECLAIM_RELEASE";

    let Some(ready) = std::env::var_os(READY_ENV).map(PathBuf::from) else {
        return Ok(());
    };
    let release = std::env::var_os(RELEASE_ENV)
        .map(PathBuf::from)
        .ok_or_else(|| anyhow::anyhow!("{RELEASE_ENV} is required with {READY_ENV}"))?;
    std::fs::write(&ready, b"reclaim lease held")?;
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    while !release.exists() {
        if std::time::Instant::now() >= deadline {
            anyhow::bail!("timed out waiting for {}", release.display());
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    Ok(())
}

fn try_reclaim_lease(lock_path: &Path) -> anyhow::Result<Option<ReclaimLease>> {
    let path = reclaim_lease_path_for(lock_path);
    let file = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&path)?;
    if try_native_exclusive_lock(&file)? {
        Ok(Some(ReclaimLease { _file: file }))
    } else {
        Ok(None)
    }
}

fn reclaim_lease_path_for(lock_path: &Path) -> PathBuf {
    let name = lock_path
        .file_name()
        .map(|name| name.to_string_lossy().into_owned())
        .unwrap_or_default();
    lock_path.with_file_name(format!("{name}.reclaim"))
}

#[cfg(unix)]
fn try_native_exclusive_lock(file: &std::fs::File) -> std::io::Result<bool> {
    use std::os::fd::AsRawFd as _;

    loop {
        // SAFETY: `file` owns a valid descriptor for the duration of this call;
        // LOCK_NB makes contention observable without blocking the retry loop.
        let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
        if result == 0 {
            return Ok(true);
        }
        let error = std::io::Error::last_os_error();
        if error.kind() == std::io::ErrorKind::Interrupted {
            continue;
        }
        if error.kind() == std::io::ErrorKind::WouldBlock {
            return Ok(false);
        }
        return Err(error);
    }
}

#[cfg(windows)]
fn try_native_exclusive_lock(file: &std::fs::File) -> std::io::Result<bool> {
    use std::os::windows::io::AsRawHandle as _;
    use windows_sys::Win32::Foundation::{ERROR_LOCK_VIOLATION, HANDLE};
    use windows_sys::Win32::Storage::FileSystem::LockFile;

    // SAFETY: `file` owns a valid Windows file handle for the call's duration.
    // LockFile is non-blocking and the held handle releases the byte-range lock
    // on Drop/process death.
    let handle = file.as_raw_handle() as HANDLE;
    let result = unsafe { LockFile(handle, 0, 0, 1, 0) };
    if result != 0 {
        return Ok(true);
    }
    let error = std::io::Error::last_os_error();
    if error.raw_os_error() == Some(ERROR_LOCK_VIOLATION as i32) {
        return Ok(false);
    }
    Err(error)
}

fn reclaimable_lock(lock_path: &Path) -> bool {
    #[cfg(test)]
    let now = PINNED_NOW
        .with(std::cell::Cell::get)
        .unwrap_or_else(SystemTime::now);
    #[cfg(not(test))]
    let now = SystemTime::now();
    reclaimable_lock_at(lock_path, now)
}

/// [`reclaimable_lock`] with the clock injected: the age comparisons read
/// `now`, so a test states the lock's age outright instead of racing the
/// grace against scheduler latency. Production passes `SystemTime::now()`.
fn reclaimable_lock_at(lock_path: &Path, now: SystemTime) -> bool {
    let Ok(body) = std::fs::read_to_string(lock_path) else {
        return false;
    };
    let age = || {
        std::fs::metadata(lock_path)
            .and_then(|metadata| metadata.modified())
            .ok()
            .and_then(|modified| now.duration_since(modified).ok())
    };
    match LockOwner::decode(&body) {
        Some(owner) => !crate::store::pid_is_alive(owner.pid),
        // This version publishes the lock's body atomically (see
        // `publish_lock`) and never observes it empty, so an empty file can
        // only be a crash artifact left by an OLDER build's `create_new` +
        // `write_all` — but that older build's writer is ALSO empty for the
        // instant between those two calls, so an empty lock isn't
        // immediately a dead one: give it the same ~2s contention budget a
        // live writer gets before calling it dead. A non-empty but
        // undecodable body has unknown provenance and keeps the
        // conservative 30s legacy age fallback.
        None if body.is_empty() => age().is_some_and(|age| age > EMPTY_LOCK_GRACE),
        None => age().is_some_and(|age| age > LEGACY_LOCK_STALE),
    }
}

/// A destination resolved exactly once. Lock selection, staging, and commit
/// all use this bound path so a symlink retarget cannot move a transaction to a
/// different file after its lock was acquired.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ResolvedPath {
    path: PathBuf,
}

impl ResolvedPath {
    /// Resolve `path` once, failing closed for dangling symlinks.
    pub fn resolve(path: &Path) -> anyhow::Result<Self> {
        Ok(Self {
            path: resolve_stable_path(path)?,
        })
    }

    #[must_use]
    pub fn as_path(&self) -> &Path {
        &self.path
    }

    #[must_use]
    pub fn lock_path(&self) -> PathBuf {
        lock_path_for(&self.path)
    }

    pub fn stage(&self, bytes: &[u8]) -> anyhow::Result<PathBuf> {
        stage_file_at(&self.path, bytes, None, false)
    }

    pub fn stage_private(&self, bytes: &[u8]) -> anyhow::Result<PathBuf> {
        stage_file_at(&self.path, bytes, None, true)
    }

    pub fn stage_with_permissions(
        &self,
        bytes: &[u8],
        permissions: Option<&std::fs::Permissions>,
        private_if_new: bool,
    ) -> anyhow::Result<PathBuf> {
        stage_file_at(&self.path, bytes, permissions, private_if_new)
    }

    /// Remove sibling staging files left by writers whose owner process is no
    /// longer alive. Call only while holding this destination's canonical lock.
    /// Live, malformed, or PID-reused stages are preserved fail-closed.
    pub fn cleanup_abandoned_stages(&self) -> anyhow::Result<usize> {
        let Some(parent) = self
            .path
            .parent()
            .filter(|path| !path.as_os_str().is_empty())
        else {
            return Ok(0);
        };
        let file_name = self
            .path
            .file_name()
            .ok_or_else(|| anyhow::anyhow!("{} has no file name", self.path.display()))?
            .to_string_lossy();
        let prefix = format!(".{file_name}.");
        let mut removed = 0;
        for entry in std::fs::read_dir(parent)? {
            let entry = entry?;
            let name = entry.file_name();
            let name = name.to_string_lossy();
            let Some(owner_and_nonce) = name
                .strip_prefix(&prefix)
                .and_then(|name| name.strip_suffix(".tmp"))
            else {
                continue;
            };
            let Some(pid) = owner_and_nonce
                .split_once('-')
                .and_then(|(pid, _)| pid.parse::<i64>().ok())
                .filter(|pid| *pid > 0)
            else {
                continue;
            };
            if crate::store::pid_is_alive(pid) {
                continue;
            }
            match std::fs::remove_file(entry.path()) {
                Ok(()) => removed += 1,
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => return Err(error.into()),
            }
        }
        if removed > 0 {
            sync_parent(&self.path)?;
        }
        Ok(removed)
    }

    pub fn durable_replace(&self, staged: &Path) -> Result<(), DurableReplaceError> {
        self.durable_replace_with_sync(staged, sync_parent_io)
    }

    /// Replace using an injected parent-sync operation while preserving the
    /// before/after-commit error contract. Transaction adapters use this only
    /// for deterministic durability failpoints; ordinary callers use
    /// [`Self::durable_replace`].
    pub fn durable_replace_with_sync(
        &self,
        staged: &Path,
        sync: impl FnOnce(&Path) -> std::io::Result<()>,
    ) -> Result<(), DurableReplaceError> {
        replace_file(staged, &self.path)
            .map_err(|source| DurableReplaceError::before_commit(&self.path, source))?;
        sync(&self.path).map_err(|source| DurableReplaceError::after_commit(&self.path, source))
    }

    pub fn durable_create(&self, staged: &Path) -> anyhow::Result<()> {
        create_file_no_replace(staged, &self.path)?;
        sync_parent(&self.path)
    }

    pub fn atomic_write(&self, bytes: &[u8]) -> anyhow::Result<()> {
        let staged = self.stage(bytes)?;
        if let Err(error) = self.durable_replace(&staged) {
            let _ = std::fs::remove_file(&staged);
            return Err(error.into());
        }
        Ok(())
    }

    pub fn atomic_write_private(&self, bytes: &[u8]) -> anyhow::Result<()> {
        let staged = self.stage_private(bytes)?;
        if let Err(error) = self.durable_replace(&staged) {
            let _ = std::fs::remove_file(&staged);
            return Err(error.into());
        }
        Ok(())
    }
}

fn resolve_stable_path(path: &Path) -> anyhow::Result<PathBuf> {
    match std::fs::symlink_metadata(path) {
        Ok(_) => return Ok(std::fs::canonicalize(path)?),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(error.into()),
    }
    let parent = path.parent().filter(|p| !p.as_os_str().is_empty());
    let parent = match parent {
        Some(parent) => {
            std::fs::create_dir_all(parent)?;
            std::fs::canonicalize(parent)?
        }
        None => std::env::current_dir()?,
    };
    let name = path
        .file_name()
        .ok_or_else(|| anyhow::anyhow!("{} has no file name", path.display()))?;
    Ok(parent.join(name))
}

/// The canonical sidecar lock identity for `path`.
pub fn stable_lock_path_for(path: &Path) -> anyhow::Result<PathBuf> {
    Ok(ResolvedPath::resolve(path)?.lock_path())
}

fn stage_file_at(
    path: &Path,
    bytes: &[u8],
    permissions: Option<&std::fs::Permissions>,
    private_if_new: bool,
) -> anyhow::Result<PathBuf> {
    let file_name = path
        .file_name()
        .ok_or_else(|| anyhow::anyhow!("{} has no file name", path.display()))?
        .to_string_lossy();
    let tmp = path.with_file_name(format!(".{file_name}.{}.tmp", unique_suffix()));
    let mut options = std::fs::OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        options.mode(if private_if_new { 0o600 } else { 0o666 });
    }
    let result = (|| -> anyhow::Result<()> {
        let mut file = options.open(&tmp)?;
        if let Some(permissions) = permissions {
            file.set_permissions(permissions.clone())?;
        } else if private_if_new {
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt as _;
                file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
            }
        } else if let Ok(metadata) = std::fs::metadata(path) {
            file.set_permissions(metadata.permissions())?;
        }
        file.write_all(bytes)?;
        file.sync_all()?;
        Ok(())
    })();
    if let Err(error) = result {
        let _ = std::fs::remove_file(&tmp);
        return Err(error);
    }
    Ok(tmp)
}

/// Atomically replace `path` with a synced sibling stage, then sync the parent
/// directory so the name update itself is durable.
pub fn durable_replace(staged: &Path, path: &Path) -> anyhow::Result<()> {
    Ok(ResolvedPath::resolve(path)?.durable_replace(staged)?)
}

/// Publish a synced sibling stage only if `path` does not already exist.
/// Used for generated drop-ins whose operator-owned name must never be
/// clobbered by a racing setup process.
pub fn durable_create(staged: &Path, path: &Path) -> anyhow::Result<()> {
    ResolvedPath::resolve(path)?.durable_create(staged)
}

#[cfg(unix)]
fn create_file_no_replace(staged: &Path, destination: &Path) -> std::io::Result<()> {
    std::fs::hard_link(staged, destination)
}

#[cfg(windows)]
fn create_file_no_replace(staged: &Path, destination: &Path) -> std::io::Result<()> {
    move_file_ex(staged, destination, false)
}

#[cfg(not(windows))]
fn replace_file(staged: &Path, destination: &Path) -> std::io::Result<()> {
    std::fs::rename(staged, destination)
}

#[cfg(windows)]
fn replace_file(staged: &Path, destination: &Path) -> std::io::Result<()> {
    move_file_ex(staged, destination, true)
}

#[cfg(windows)]
fn move_file_ex(staged: &Path, destination: &Path, replace: bool) -> std::io::Result<()> {
    use std::os::windows::ffi::OsStrExt as _;
    use windows_sys::Win32::Storage::FileSystem::{
        MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH,
    };

    let source: Vec<u16> = staged.as_os_str().encode_wide().chain(Some(0)).collect();
    let destination: Vec<u16> = destination
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect();
    // SAFETY: both pointers reference NUL-terminated UTF-16 buffers valid for
    // the duration of the call. MOVEFILE_REPLACE_EXISTING provides the Windows
    // replace semantics `rename` cannot portably promise; WRITE_THROUGH is the
    // Windows equivalent of making the directory-entry update durable.
    let flags = MOVEFILE_WRITE_THROUGH
        | if replace {
            MOVEFILE_REPLACE_EXISTING
        } else {
            0
        };
    let result = unsafe { MoveFileExW(source.as_ptr(), destination.as_ptr(), flags) };
    if result == 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

/// Write `bytes` to `path` atomically and durably.
pub fn atomic_write(path: &Path, bytes: &[u8]) -> anyhow::Result<()> {
    ResolvedPath::resolve(path)?.atomic_write(bytes)
}

/// Write private `bytes` to `path` atomically and durably (0600 on Unix).
pub fn atomic_write_private(path: &Path, bytes: &[u8]) -> anyhow::Result<()> {
    ResolvedPath::resolve(path)?.atomic_write_private(bytes)
}

/// Sync the directory entry containing `path`.
pub fn sync_parent(path: &Path) -> anyhow::Result<()> {
    Ok(sync_parent_io(path)?)
}

fn sync_parent_io(path: &Path) -> std::io::Result<()> {
    let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) else {
        return Ok(());
    };
    #[cfg(unix)]
    {
        std::fs::File::open(parent)?.sync_all()?;
    }
    // Windows cannot portably FlushFileBuffers on a directory handle. The
    // durable replacement path above uses MOVEFILE_WRITE_THROUGH instead.
    #[cfg(windows)]
    let _ = parent;
    Ok(())
}

/// The conventional sidecar lock path for `path`: `<path>.lock`.
#[must_use]
pub fn lock_path_for(path: &Path) -> PathBuf {
    let name = path
        .file_name()
        .map(|name| name.to_string_lossy().into_owned())
        .unwrap_or_default();
    path.with_file_name(format!("{name}.lock"))
}

/// Collision-resistant process-local suffix suitable for staged/versioned
/// file names.  It contains no credentials or host-specific data.
#[must_use]
pub fn unique_suffix() -> String {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    let counter = UNIQUE_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("{}-{nanos}-{counter}", std::process::id())
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    /// **The twin for [`is_lock_contended`] (#1871).**
    ///
    /// The positive case is produced END TO END — the lock is really held and
    /// a second thread really runs `acquire_lock` to exhaustion. Nothing here
    /// writes the message by hand: a predicate checked against a hand-copied
    /// copy of the string it matches would pass no matter how far the two had
    /// drifted, which is the failure this whole slice is about.
    ///
    /// The negative cases are the discriminating half. "Did `acquire_lock`
    /// fail?" is NOT the question — the permission refusal is also a failure
    /// from the same function, and retrying it just fails slower.
    ///
    /// Costs the full retry budget (~2s) by construction: exhausting it is the
    /// event under test.
    /// A lock this version writes is never observed empty: `publish_lock`
    /// only makes `lock_path` exist via `hard_link` from a temp file that
    /// already carries the full `pid:nonce` body, so there is no window
    /// where the path exists with zero bytes (unlike the old direct
    /// `create_new` + `write_all` sequence).
    #[test]
    fn a_lock_acquired_by_this_version_is_never_observed_empty() {
        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("config.toml"));
        let guard = acquire_lock(&lock).unwrap();
        let body = std::fs::read_to_string(&lock).unwrap();
        assert!(
            !body.is_empty(),
            "the lock must carry its final body from the instant it becomes visible"
        );
        assert!(LockOwner::decode(&body).is_some(), "{body:?}");
        drop(guard);
    }

    #[test]
    fn contention_is_recognised_and_nothing_else_is() {
        let dir = TempDir::new().unwrap();
        let lock = dir.path().join("peers.toml.lock");
        let _held = acquire_lock(&lock).expect("take the lock");

        let refused = std::thread::scope(|scope| {
            scope
                .spawn(|| acquire_lock(&lock).expect_err("the lock is held"))
                .join()
                .expect("probe thread")
        });
        assert!(
            is_lock_contended(&refused),
            "a live holder must read as retryable contention: {refused:#}"
        );

        assert!(
            !is_lock_contended(&anyhow::anyhow!(
                "could not acquire /x — permission denied; check the lock directory permissions"
            )),
            "a permission refusal is not retryable — the predicate must not \
             simply mean `acquire_lock failed`"
        );
        assert!(!is_lock_contended(&anyhow::anyhow!(
            "no space left on device"
        )));
    }

    /// **Red for #2487.** A lock file killed between `create_new` and
    /// `write_all` is empty — no `pid:nonce` body, so `LockOwner::decode`
    /// returns `None` and reclamation falls back to the 30s legacy-age path.
    /// Within the ~2s retry budget that reads as CONTENDED — "another live
    /// process" — even though no process anywhere holds it and the body is
    /// unmistakably a crash artifact of the *current* write sequence, never a
    /// value this version writes on purpose. Reproduces the report: a stale
    /// 0-byte lock, no live owner, nothing to retry into. `acquire_lock`
    /// itself retries for ~2s (well past `EMPTY_LOCK_GRACE`), so this level
    /// grounds the end-to-end behavior; the two tests below pin the grace
    /// boundary `reclaimable_lock` actually applies.
    #[test]
    fn empty_crash_artifact_lock_is_reclaimed_not_reported_live() {
        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("config.toml"));
        // Simulates `create_new` succeeding and the process dying before
        // `write_all` — exactly what a kill/OOM/panic between those two
        // calls in `acquire_lock` leaves on disk. Freshly written, so the
        // legacy 30s age check does not consider it stale.
        std::fs::File::create(&lock).unwrap();

        let guard = acquire_lock(&lock).expect(
            "an empty lock from this version's own write sequence is a dead owner, not a live one",
        );
        drop(guard);
    }

    /// A fresh empty lock (age well under [`EMPTY_LOCK_GRACE`]) is NOT
    /// reclaimable yet — this is the safety half of the item-1 fix: a LIVE
    /// writer's own lock is also empty for the instant between its
    /// `create_new` and `write_all` (see [`publish_lock`]'s fallback path,
    /// and any older build's direct write), and a contender must not delete
    /// that window's lock out from under it.
    #[test]
    fn fresh_empty_lock_is_not_reclaimed_within_the_grace() {
        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("config.toml"));
        let file = std::fs::File::create(&lock).unwrap();
        // The age is stated, not measured: the lock's mtime is pinned and the
        // reader's clock is one tick inside the grace. The former form ("created
        // the statement before") was a bet that fewer than 200 ms passed between
        // two adjacent statements, which a loaded box does not honour.
        let created = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        file.set_modified(created).unwrap();
        assert!(
            !reclaimable_lock_at(&lock, created + EMPTY_LOCK_GRACE - Duration::from_millis(1)),
            "a fresh empty lock must not read as a dead owner yet"
        );
        assert!(
            reclaimable_lock_at(&lock, created + EMPTY_LOCK_GRACE + Duration::from_millis(1)),
            "one tick past the grace, the same empty lock is a crash artifact"
        );
    }

    /// Past [`EMPTY_LOCK_GRACE`], an empty lock IS reclaimable — the crash
    /// case this fix targets.
    #[test]
    fn aged_empty_lock_past_the_grace_is_reclaimed() {
        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("config.toml"));
        let file = std::fs::File::create(&lock).unwrap();
        file.set_modified(
            std::time::SystemTime::now() - EMPTY_LOCK_GRACE - Duration::from_millis(50),
        )
        .unwrap();
        assert!(
            reclaimable_lock(&lock),
            "an empty lock older than the grace is a crash artifact, not a live writer"
        );
    }

    #[test]
    fn atomic_write_replaces_and_leaves_no_tmp() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("f.toml");
        atomic_write(&path, b"one").unwrap();
        atomic_write(&path, b"two").unwrap();
        assert_eq!(std::fs::read(&path).unwrap(), b"two");
        assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 1);
    }

    /// #2525 review item 1: a link failure other than `AlreadyExists` — the
    /// EPERM a vfat/exFAT/FUSE(sshfs) target maps to `PermissionDenied`, or an
    /// Uncategorized Windows FAT/SMB error — must fall back to the old
    /// create_new path, not be treated as contention or fail outright. Seam
    /// (`publish_lock_with_link`) stands in for the real mount so this stays
    /// in the fully-mocked unit tier.
    #[test]
    fn hard_link_failure_other_than_unsupported_falls_back_to_create_new() {
        let dir = TempDir::new().unwrap();
        let lock_path = dir.path().join("config.toml.lock");
        publish_lock_with_link(&lock_path, "owner-body", |_src, _dst| {
            Err(std::io::Error::from(std::io::ErrorKind::PermissionDenied))
        })
        .expect("falls back instead of surfacing the injected link error");
        assert_eq!(std::fs::read_to_string(&lock_path).unwrap(), "owner-body");
    }

    /// #2525 review item 2: a crash between the temp-file write and the
    /// `hard_link`/fallback publish leaves a `<lock>.*.tmp` orphan behind.
    /// Once past the contention retry budget it is safe to call dead and
    /// must be swept; a fresh one still within budget (an in-flight writer)
    /// must survive.
    #[test]
    fn sweep_lock_tmp_orphans_removes_only_ones_past_the_retry_budget() {
        let dir = TempDir::new().unwrap();
        let lock_path = dir.path().join("config.toml.lock");
        let stale = dir.path().join("config.toml.lock.stale-orphan.tmp");
        std::fs::write(&stale, b"orphan").unwrap();
        // Aged explicitly, not by sleeping, so scheduler latency and mtime
        // granularity cannot decide the outcome.
        std::fs::File::options()
            .write(true)
            .open(&stale)
            .unwrap()
            .set_modified(std::time::SystemTime::now() - Duration::from_secs(60))
            .unwrap();
        let fresh = dir.path().join("config.toml.lock.fresh-orphan.tmp");
        std::fs::write(&fresh, b"orphan").unwrap();

        sweep_lock_tmp_orphans_older_than(&lock_path, Duration::from_secs(30));

        assert!(
            !stale.exists(),
            "orphan past the retry budget must be swept"
        );
        assert!(
            fresh.exists(),
            "orphan within the retry budget must survive"
        );
    }

    #[test]
    #[ignore = "real filesystem cleanup acceptance; run in mcp-import-real workflow"]
    #[serial_test::serial(real_fs)]
    fn abandoned_stage_cleanup_removes_only_dead_owner_for_destination() {
        let dir = TempDir::new().unwrap();
        let destination = ResolvedPath::resolve(&dir.path().join("config.toml")).unwrap();
        let dead = dir.path().join(".config.toml.4294967295-dead-0.tmp");
        let live = dir
            .path()
            .join(format!(".config.toml.{}-live-0.tmp", std::process::id()));
        let unrelated = dir.path().join(".mcp.toml.4294967295-dead-0.tmp");
        for path in [&dead, &live, &unrelated] {
            std::fs::write(path, b"stage").unwrap();
        }

        let _guard = acquire_lock(&destination.lock_path()).unwrap();
        assert_eq!(destination.cleanup_abandoned_stages().unwrap(), 1);
        assert!(!dead.exists());
        assert!(live.exists());
        assert!(unrelated.exists());
    }

    #[test]
    fn a_lock_excludes_a_second_holder_until_dropped() {
        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("f.toml"));
        let guard = acquire_lock(&lock).unwrap();
        assert!(lock.exists());
        drop(guard);
        assert!(!lock.exists());
        let _again = acquire_lock(&lock).unwrap();
    }

    /// Real-filesystem grounding for immediate abandoned-owner recovery.
    #[ignore = "real-resource: weekly/release tier; touches the filesystem"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn dead_owner_is_reclaimed_without_waiting_for_age_threshold() {
        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("f.toml"));
        // u32::MAX cannot be a live Unix pid and OpenProcess rejects it on
        // Windows; keep the fixture positive so it never acquires process-group
        // semantics (`kill(0/-1, 0)`).
        std::fs::write(&lock, "4294967295:abandoned\n").unwrap();
        let _guard = acquire_lock(&lock).unwrap();
        let owner = LockOwner::decode(&std::fs::read_to_string(&lock).unwrap()).unwrap();
        assert_eq!(owner.pid, i64::from(std::process::id()));
    }

    /// Real-filesystem grounding for canonical lock identity selection.
    #[ignore = "real-resource: weekly/release tier; touches the filesystem"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn aliases_share_one_stable_lock_identity() {
        let dir = TempDir::new().unwrap();
        let destination = dir.path().join("config.toml");
        std::fs::write(&destination, "x").unwrap();
        let relative = dir.path().join(".").join("config.toml");
        assert_eq!(
            stable_lock_path_for(&destination).unwrap(),
            stable_lock_path_for(&relative).unwrap()
        );
    }

    #[cfg(unix)]
    /// Real-filesystem grounding for private replacement permissions.
    #[ignore = "real-resource: weekly/release tier; touches the filesystem"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn private_replacement_tightens_existing_permissions() {
        use std::os::unix::fs::PermissionsExt as _;

        let dir = TempDir::new().unwrap();
        let path = dir.path().join("token.age");
        std::fs::write(&path, b"old").unwrap();
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();

        atomic_write_private(&path, b"secret").unwrap();

        assert_eq!(
            std::fs::metadata(path).unwrap().permissions().mode() & 0o777,
            0o600
        );
    }

    #[cfg(unix)]
    /// Real-filesystem grounding for fail-closed dangling-symlink handling.
    #[ignore = "real-resource: weekly/release tier; touches the filesystem"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn dangling_symlink_is_rejected_instead_of_replaced() {
        use std::os::unix::fs::symlink;

        let dir = TempDir::new().unwrap();
        let target = dir.path().join("missing-target");
        let link = dir.path().join("config.toml");
        symlink(&target, &link).unwrap();

        let error = atomic_write(&link, b"body").unwrap_err();

        assert!(std::fs::symlink_metadata(&link)
            .unwrap()
            .file_type()
            .is_symlink());
        assert!(!target.exists());
        assert!(
            error.to_string().contains("No such file") || error.to_string().contains("not find")
        );
    }

    #[cfg(unix)]
    /// Real-filesystem grounding for the transaction's bound destination.
    #[ignore = "real-resource: weekly/release tier; retargets a filesystem symlink"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn resolved_destination_cannot_be_retargeted_mid_transaction() {
        use std::os::unix::fs::symlink;

        let dir = TempDir::new().unwrap();
        let first = dir.path().join("first.toml");
        let second = dir.path().join("second.toml");
        let link = dir.path().join("config.toml");
        std::fs::write(&first, b"first-old").unwrap();
        std::fs::write(&second, b"second-old").unwrap();
        symlink(&first, &link).unwrap();

        let destination = ResolvedPath::resolve(&link).unwrap();
        let _guard = acquire_lock(&destination.lock_path()).unwrap();
        let staged = destination.stage(b"first-new").unwrap();
        std::fs::remove_file(&link).unwrap();
        symlink(&second, &link).unwrap();

        destination.durable_replace(&staged).unwrap();

        assert_eq!(std::fs::read(&first).unwrap(), b"first-new");
        assert_eq!(std::fs::read(&second).unwrap(), b"second-old");
        assert_eq!(
            std::fs::canonicalize(&link).unwrap(),
            std::fs::canonicalize(second).unwrap()
        );
    }

    /// Real-filesystem grounding for platform replacement semantics.
    #[ignore = "real-resource: weekly/release tier; touches the filesystem"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn replacement_overwrites_an_existing_destination() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("config.toml");
        atomic_write(&path, b"old").unwrap();
        atomic_write(&path, b"new").unwrap();
        assert_eq!(std::fs::read(path).unwrap(), b"new");
    }

    /// Real-filesystem grounding for the commit-state error contract used by
    /// setup transactions: the replacement is visible when parent sync fails.
    #[ignore = "real-resource: weekly/release tier; touches the filesystem"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn sync_failure_after_replacement_reports_committed() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("config.toml");
        let destination = ResolvedPath::resolve(&path).unwrap();
        let staged = destination.stage(b"new").unwrap();

        let error = destination
            .durable_replace_with_sync(&staged, |_| {
                Err(std::io::Error::other("injected parent sync failure"))
            })
            .unwrap_err();

        assert!(error.committed());
        assert_eq!(std::fs::read(path).unwrap(), b"new");
        assert!(error.to_string().contains("could not durably sync"));
    }

    /// Real-process grounding for the owner-aware stale-lock contract. The
    /// child is killed (so `LockGuard::drop` cannot run); the parent must then
    /// reclaim its lock immediately on both Unix and Windows.
    #[ignore = "real-resource: weekly/release tier; spawns and kills a process"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn killed_process_lock_is_reclaimed() {
        const CHILD_ENV: &str = "NEWT_ATOMIC_LOCK_CHILD";
        const LOCK_ENV: &str = "NEWT_ATOMIC_LOCK_PATH";
        const READY_ENV: &str = "NEWT_ATOMIC_LOCK_READY";

        if std::env::var_os(CHILD_ENV).is_some() {
            let lock = PathBuf::from(std::env::var_os(LOCK_ENV).expect("child lock path"));
            let ready = PathBuf::from(std::env::var_os(READY_ENV).expect("child ready path"));
            let _guard = acquire_lock(&lock).expect("child acquires lock");
            std::fs::write(ready, b"ready").expect("child publishes readiness");
            loop {
                std::thread::sleep(Duration::from_secs(1));
            }
        }

        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("config.toml"));
        let ready = dir.path().join("ready");
        let mut child = std::process::Command::new(std::env::current_exe().unwrap())
            .arg("--exact")
            .arg("atomic_fs::tests::killed_process_lock_is_reclaimed")
            .arg("--ignored")
            .arg("--nocapture")
            .env(CHILD_ENV, "1")
            .env(LOCK_ENV, &lock)
            .env(READY_ENV, &ready)
            .spawn()
            .unwrap();

        let deadline = std::time::Instant::now() + crate::test_guard::HANG_GUARD;
        while !ready.exists() && std::time::Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(20));
        }
        if !ready.exists() {
            let _ = child.kill();
            let _ = child.wait();
            panic!("lock-holder child did not become ready");
        }
        // Readiness proves existence, not freshness: the parent may have been
        // descheduled before observing it. Pin to the lock's own timestamp so
        // its age is exactly zero regardless of any delay before this read.
        let lock_is_fresh_at = std::fs::metadata(&lock).unwrap().modified().unwrap();
        child.kill().unwrap();
        let exit_status = child.wait().unwrap();

        // Count reclaim attempts with the lock's age pinned to zero. A dead
        // owner must be reclaimed on the first attempt; requiring the legacy
        // age threshold must fail even if real scheduling delays exceed it.
        reset_reclaim_attempts_for_test();
        let _unpin = pin_now_for_test(lock_is_fresh_at);
        let _guard = acquire_lock(&lock).expect("dead owner's lock is reclaimable");
        assert_eq!(
            reclaim_attempts_for_test(),
            1,
            "owner-aware recovery must reclaim on its first attempt, not fall \
             through to the legacy age-based retry loop"
        );
        // Keep `Child` (and therefore its Windows process handle) alive through
        // reclamation. An exited Windows process remains openable while this
        // handle exists, so this specifically guards against treating a
        // successful `OpenProcess` as proof of liveness.
        assert_eq!(child.try_wait().unwrap(), Some(exit_status));
    }

    /// Real-process grounding for serialized stale-lock takeover. The first
    /// reclaimer is paused while it holds the recovery lease, guaranteeing the
    /// second contender races the same abandoned generation. Both contenders
    /// must subsequently enter the protected section without overlapping.
    #[ignore = "real-resource: weekly/release tier; spawns and kills processes"]
    #[serial_test::serial(real_fs)]
    #[test]
    fn two_reclaimers_cannot_delete_a_new_lock_generation() {
        const ROLE_ENV: &str = "NEWT_ATOMIC_RACE_ROLE";
        const LOCK_ENV: &str = "NEWT_ATOMIC_RACE_LOCK";
        const READY_ENV: &str = "NEWT_ATOMIC_RACE_READY";
        const STARTED_ENV: &str = "NEWT_ATOMIC_RACE_STARTED";
        const CRITICAL_ENV: &str = "NEWT_ATOMIC_RACE_CRITICAL";

        if let Some(role) = std::env::var_os(ROLE_ENV) {
            let lock = PathBuf::from(std::env::var_os(LOCK_ENV).expect("child lock path"));
            let ready = PathBuf::from(std::env::var_os(READY_ENV).expect("child ready path"));
            if role == "holder" {
                let _guard = acquire_lock(&lock).expect("holder acquires lock");
                std::fs::write(ready, b"holder ready").expect("holder publishes readiness");
                loop {
                    std::thread::sleep(Duration::from_secs(1));
                }
            }

            let started =
                PathBuf::from(std::env::var_os(STARTED_ENV).expect("reclaimer started path"));
            let critical = PathBuf::from(
                std::env::var_os(CRITICAL_ENV).expect("reclaimer critical-section path"),
            );
            std::fs::write(started, role.as_encoded_bytes())
                .expect("reclaimer publishes readiness");
            let guard = acquire_lock(&lock).expect("reclaimer acquires lock");
            let marker = std::fs::OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(&critical)
                .expect("reclaimers never overlap their protected sections");
            std::thread::sleep(Duration::from_millis(250));
            drop(marker);
            std::fs::remove_file(&critical).expect("remove critical-section marker");
            drop(guard);
            std::fs::write(ready, role.as_encoded_bytes()).expect("reclaimer publishes completion");
            return;
        }

        fn wait_for(path: &Path, description: &str) {
            let deadline = std::time::Instant::now() + Duration::from_secs(10);
            while !path.exists() && std::time::Instant::now() < deadline {
                std::thread::sleep(Duration::from_millis(10));
            }
            assert!(path.exists(), "timed out waiting for {description}");
        }

        fn wait_for_child(child: &mut std::process::Child, description: &str) {
            let deadline = std::time::Instant::now() + Duration::from_secs(10);
            loop {
                if let Some(status) = child.try_wait().expect("poll child") {
                    assert!(status.success(), "{description} failed with {status}");
                    return;
                }
                if std::time::Instant::now() >= deadline {
                    let _ = child.kill();
                    let _ = child.wait();
                    panic!("timed out waiting for {description}");
                }
                std::thread::sleep(Duration::from_millis(10));
            }
        }

        let dir = TempDir::new().unwrap();
        let lock = lock_path_for(&dir.path().join("config.toml"));
        let holder_ready = dir.path().join("holder-ready");
        let first_started = dir.path().join("first-started");
        let second_started = dir.path().join("second-started");
        let first_done = dir.path().join("first-done");
        let second_done = dir.path().join("second-done");
        let reclaimer_ready = dir.path().join("reclaimer-lease-held");
        let reclaimer_release = dir.path().join("reclaimer-release");
        let critical = dir.path().join("critical");
        let executable = std::env::current_exe().unwrap();
        let test_name = "atomic_fs::tests::two_reclaimers_cannot_delete_a_new_lock_generation";

        let mut holder = std::process::Command::new(&executable)
            .arg("--exact")
            .arg(test_name)
            .arg("--ignored")
            .arg("--nocapture")
            .env(ROLE_ENV, "holder")
            .env(LOCK_ENV, &lock)
            .env(READY_ENV, &holder_ready)
            .spawn()
            .unwrap();
        wait_for(&holder_ready, "holder readiness");
        holder.kill().unwrap();
        holder.wait().unwrap();

        let mut first = std::process::Command::new(&executable)
            .arg("--exact")
            .arg(test_name)
            .arg("--ignored")
            .arg("--nocapture")
            .env(ROLE_ENV, "first")
            .env(LOCK_ENV, &lock)
            .env(READY_ENV, &first_done)
            .env(STARTED_ENV, &first_started)
            .env(CRITICAL_ENV, &critical)
            .env("NEWT_ATOMIC_RECLAIM_READY", &reclaimer_ready)
            .env("NEWT_ATOMIC_RECLAIM_RELEASE", &reclaimer_release)
            .spawn()
            .unwrap();
        wait_for(&first_started, "first reclaimer start");
        wait_for(&reclaimer_ready, "first reclaimer lease");

        let mut second = std::process::Command::new(&executable)
            .arg("--exact")
            .arg(test_name)
            .arg("--ignored")
            .arg("--nocapture")
            .env(ROLE_ENV, "second")
            .env(LOCK_ENV, &lock)
            .env(READY_ENV, &second_done)
            .env(STARTED_ENV, &second_started)
            .env(CRITICAL_ENV, &critical)
            .spawn()
            .unwrap();
        wait_for(&second_started, "second reclaimer start");
        std::fs::write(&reclaimer_release, b"release").unwrap();

        wait_for_child(&mut first, "first reclaimer");
        wait_for_child(&mut second, "second reclaimer");
        assert!(first_done.exists());
        assert!(second_done.exists());
        assert!(!critical.exists());
    }
}