haematite 0.6.1

//! A4: cross-process data-directory writer lock.
//!
//! One haematite data directory admits at most ONE live writer process; a
//! second writer running its own shard actors over the same WAL files would
//! corrupt them. [`DataDirLock`] closes that hole with an EXCLUSIVE advisory
//! OS lock (`flock` on Unix, `LockFileEx` on Windows, via the stable
//! `std::fs::File` locking API) over `<data_dir>/writer.lock`, acquired
//! non-blocking at `Database::create`/`Database::open` and held for the
//! lifetime of the `Database` handle.
//!
//! Semantics (see docs/design/LOCKFILE-SEMANTICS.md for the full note):
//!
//! - Contention never blocks and never corrupts: a second writer fails LOUDLY
//!   at open with the distinct [`super::DatabaseError::DataDirLocked`].
//! - There is no staleness story to manage: advisory locks belong to the open
//!   file description and the kernel releases them on process death (`kill -9`
//!   included). A crashed writer leaves only an inert, unlocked, content-free
//!   lockfile whose presence means nothing.
//! - The lockfile carries NO content (deliberately no PID — PID files lie
//!   after crashes and PID reuse; the OS lock state is the only truth).
//! - Read-only observers ([`super::ReadOnlyDatabase`]) take NO lock at all:
//!   they are safe by construction and must never be able to deny startup to
//!   the legitimate writer.
//! - Drop does NOT issue an explicit unlock: release happens when the last
//!   handle to the open file description closes. This keeps the planned haem
//!   double-fork daemon sound — a parent's fd close never yanks the lock out
//!   from under a forked child, whereas an explicit `LOCK_UN` would.

use std::fmt;
use std::fs::{self, File, OpenOptions, TryLockError};
use std::io;
use std::path::{Path, PathBuf};

use super::DatabaseError;

/// The writer lockfile name inside a database data directory.
pub(super) const LOCK_FILE: &str = "writer.lock";

/// Exclusive cross-process writer lock over one database data directory.
///
/// Constructed by [`DataDirLock::acquire`]; the OS lock is held exactly as
/// long as this value (and any file descriptions it forked to) lives. Dropping
/// it — or the process dying — releases the lock; the lockfile itself is left
/// behind as an inert, content-free anchor.
pub struct DataDirLock {
    /// The open, locked lockfile handle. Held, never read: closing it is what
    /// releases the advisory lock.
    file: File,
    /// The lockfile path, kept for diagnostics.
    lock_path: PathBuf,
    /// Whether THIS acquisition created the anchor — causal (`create_new`
    /// succeeded), never inferred from before/after probes. The vacuum's
    /// report-only transparency rule depends on this being exact.
    created_anchor: bool,
}

impl fmt::Debug for DataDirLock {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DataDirLock")
            .field("lock_path", &self.lock_path)
            .field("file", &self.file)
            .field("created_anchor", &self.created_anchor)
            .finish()
    }
}

impl DataDirLock {
    /// Acquire the exclusive writer lock for `data_dir`, without blocking.
    ///
    /// Opens (creating if absent, never truncating) `<data_dir>/writer.lock`
    /// and takes an exclusive advisory lock on it. A lock already held by any
    /// live process — including this one, through a different `Database`
    /// handle — fails immediately with [`LockError::AlreadyLocked`].
    ///
    /// ANCHOR DISCIPLINE: the anchor must be a REGULAR FILE directly in the
    /// data dir. A symlinked `writer.lock` is refused
    /// ([`LockError::AnchorNotARegularFile`]) BEFORE any create-capable
    /// open — `O_CREAT` follows a dangling link, so opening through one
    /// would mint an inode OUTSIDE the directory this lock protects.
    /// Creation is established causally (`create_new`), never inferred from
    /// before/after probes, so [`Self::created_anchor`] is exact. (The
    /// lstat-then-open pair is not atomic against an adversary racing
    /// replacements in the same directory; the threat model is damaged or
    /// mis-tooled directories, not hostile concurrent writers — which the
    /// lock itself cannot defend against either.)
    pub fn acquire(data_dir: &Path) -> Result<Self, LockError> {
        let lock_path = data_dir.join(LOCK_FILE);
        let mut attempts = 0_u32;
        let (file, created_anchor) = loop {
            attempts += 1;
            if attempts > 4 {
                return Err(LockError::Io {
                    lock_path,
                    error: io::Error::other(
                        "writer.lock flapped between present and absent across repeated \
                         attempts; another process is churning the anchor",
                    ),
                    created_anchor: false,
                });
            }
            match fs::symlink_metadata(&lock_path) {
                Ok(meta) if meta.file_type().is_file() => {
                    match OpenOptions::new().read(true).write(true).open(&lock_path) {
                        Ok(file) => break (file, false),
                        // Unlinked between lstat and open: re-evaluate.
                        Err(error) if error.kind() == io::ErrorKind::NotFound => {}
                        Err(error) => {
                            return Err(LockError::Io {
                                lock_path,
                                error,
                                created_anchor: false,
                            });
                        }
                    }
                }
                Ok(_meta) => return Err(LockError::AnchorNotARegularFile { lock_path }),
                Err(error) if error.kind() == io::ErrorKind::NotFound => {
                    match OpenOptions::new()
                        .read(true)
                        .write(true)
                        .create_new(true)
                        .open(&lock_path)
                    {
                        // `create_new` succeeded ⇒ THIS call made the anchor.
                        Ok(file) => break (file, true),
                        // Someone else created it first: re-evaluate (it may
                        // now be a file — or a link, which must refuse).
                        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
                        Err(error) => {
                            return Err(LockError::Io {
                                lock_path,
                                error,
                                created_anchor: false,
                            });
                        }
                    }
                }
                Err(error) => {
                    return Err(LockError::Io {
                        lock_path,
                        error,
                        created_anchor: false,
                    });
                }
            }
        };
        #[cfg(test)]
        if FAIL_NEXT_TRY_LOCK.with(|flag| flag.replace(false)) {
            return Err(LockError::Io {
                lock_path,
                error: io::Error::other("injected post-open lock failure (test seam)"),
                created_anchor,
            });
        }
        #[cfg(test)]
        if CONTEND_NEXT_TRY_LOCK.with(|flag| flag.replace(false)) {
            // Simulate the create-then-lose interleaving: a competitor opens
            // and locks the (now existing) anchor before OUR try_lock runs.
            if let Ok(competitor) = OpenOptions::new().read(true).write(true).open(&lock_path)
                && competitor.try_lock().is_ok()
            {
                CONTENDERS.with(|held| held.borrow_mut().push(competitor));
            }
        }
        match file.try_lock() {
            Ok(()) => Ok(Self {
                file,
                lock_path,
                created_anchor,
            }),
            Err(TryLockError::WouldBlock) => Err(LockError::AlreadyLocked {
                lock_path,
                created_anchor,
            }),
            Err(error) => Err(LockError::Io {
                lock_path,
                error: error.into(),
                created_anchor,
            }),
        }
    }

    /// Whether THIS acquisition created the anchor file (causal, exact).
    pub const fn created_anchor(&self) -> bool {
        self.created_anchor
    }
}

#[cfg(test)]
thread_local! {
    static FAIL_NEXT_TRY_LOCK: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
    static CONTEND_NEXT_TRY_LOCK: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
    static CONTENDERS: std::cell::RefCell<Vec<File>> = const { std::cell::RefCell::new(Vec::new()) };
}

/// Test seam: make the NEXT `acquire` on this thread fail at the lock
/// syscall AFTER the create-capable open — the window where an anchor can
/// exist without a lock. Lets the vacuum pin its disclosure of that write.
#[cfg(test)]
pub(super) fn fail_next_try_lock() {
    FAIL_NEXT_TRY_LOCK.with(|flag| flag.set(true));
}

/// Test seam: make the NEXT `acquire` on this thread LOSE the lock race to a
/// competitor that grabs the OS lock between open and `try_lock` — the
/// create-then-lose interleaving whose disclosure r4-M5 requires.
#[cfg(test)]
pub(super) fn contend_next_try_lock() {
    CONTEND_NEXT_TRY_LOCK.with(|flag| flag.set(true));
}

/// Release any competitor handles the contention seam is holding.
#[cfg(test)]
pub(super) fn release_contenders() {
    CONTENDERS.with(|held| held.borrow_mut().clear());
}

/// Errors raised while acquiring the data-dir writer lock.
#[derive(Debug)]
pub enum LockError {
    /// Another live writer (process or in-process `Database` handle) holds the
    /// exclusive lock. The loud second-writer failure.
    AlreadyLocked {
        lock_path: PathBuf,
        /// Whether THIS losing acquisition had already created the anchor
        /// (create-then-lose-the-race interleaving) — the causal fact must
        /// survive every exit so disclosure is never dropped.
        created_anchor: bool,
    },
    /// `writer.lock` exists but is not a regular file (symlink or special
    /// entry). Refused BEFORE any create-capable open: `O_CREAT` through a
    /// dangling link would mint an inode outside the data dir.
    AnchorNotARegularFile { lock_path: PathBuf },
    /// The lockfile could not be opened/created, or the lock syscall failed
    /// for an I/O reason distinct from contention.
    Io {
        lock_path: PathBuf,
        error: io::Error,
        /// Whether the failed acquisition had already CREATED the anchor
        /// (causal — `create_new` succeeded before the failure).
        created_anchor: bool,
    },
}

impl fmt::Display for LockError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::AlreadyLocked { lock_path, .. } => write!(
                formatter,
                "data dir writer lock at {} is held by another live writer",
                lock_path.display()
            ),
            Self::AnchorNotARegularFile { lock_path } => write!(
                formatter,
                "writer.lock at {} is not a regular file (symlink or special entry); the \
                 advisory anchor must be a plain file in the data dir — remove or replace \
                 it with a plain empty file, then re-run",
                lock_path.display()
            ),
            Self::Io {
                lock_path, error, ..
            } => write!(
                formatter,
                "failed to acquire data dir writer lock at {}: {error}",
                lock_path.display()
            ),
        }
    }
}

impl std::error::Error for LockError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io { error, .. } => Some(error),
            Self::AlreadyLocked { .. } | Self::AnchorNotARegularFile { .. } => None,
        }
    }
}

impl From<LockError> for DatabaseError {
    fn from(error: LockError) -> Self {
        match error {
            LockError::AlreadyLocked { lock_path, .. } => Self::DataDirLocked { lock_path },
            LockError::AnchorNotARegularFile { lock_path } => Self::LockFileIo {
                error: io::Error::other(
                    "writer.lock is not a regular file (symlink or special entry); the \
                     advisory anchor must be a plain file in the data dir",
                ),
                lock_path,
            },
            LockError::Io {
                lock_path, error, ..
            } => Self::LockFileIo { lock_path, error },
        }
    }
}