tephra 0.4.0

//! The single-writer lock on a data directory.
//!
//! A data directory takes one writer at a time. That is enforced in two halves, because
//! neither alone is sufficient:
//!
//! - **Across processes**, a POSIX record lock (`fcntl` `F_SETLK`) on a `LOCK` file. Record
//!   locks are owned by the *process*, not by the open file description, which is exactly the
//!   property needed here: `flock` and OFD locks both ride the descriptor, so a `fork` (any
//!   `Command::spawn`, between the fork and the exec) duplicates them, and a process that
//!   spawns a subprocess while holding one can then be refused its own directory after a
//!   clean shutdown, by a lock it cannot break because nothing still names it. Record locks
//!   are simply not inherited.
//! - **Within one process**, a registry of locked directories. The flip side of process
//!   ownership is that a second `F_SETLK` from the same process succeeds rather than
//!   conflicting, so the kernel cannot tell a second `SegmentSet` here apart from the first.
//!   The registry supplies that half. This is the same split SQLite and RocksDB use.
//!
//! The record lock also carries a footgun worth naming: closing *any* descriptor this process
//! holds on the lock file releases the process's locks on it. Nothing else in the engine ever
//! opens `LOCK`, and the registry keeps a second opener from getting that far.

use std::collections::HashSet;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Write as _};
use std::path::{Path, PathBuf};
use std::process;
use std::str;
use std::sync::{Mutex, OnceLock};

#[cfg(unix)]
use std::os::unix::fs::FileExt;

/// Advisory single-writer lock file, held for as long as a read-write set lives. Its contents
/// are a pid, used only to make the contended error actionable.
pub(crate) const LOCK_FILE: &str = "LOCK";

/// Directories this process currently holds, keyed by canonical path.
fn held() -> &'static Mutex<HashSet<PathBuf>> {
    static HELD: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
    HELD.get_or_init(|| Mutex::new(HashSet::new()))
}

/// Why a directory could not be locked.
pub(crate) enum LockFailure {
    /// Another handle, in this process or another, already holds it.
    Contended {
        path: PathBuf,
        /// The pid the holder recorded, when it could be read. Advisory and possibly stale.
        holder: Option<u32>,
    },
    /// The lock file itself could not be opened or locked.
    Io { path: PathBuf, source: io::Error },
}

/// A held directory lock. Dropping it releases both halves.
#[derive(Debug)]
pub(crate) struct DirLock {
    /// Held open for the lock's lifetime: closing it releases the record lock.
    _file: File,
    key: PathBuf,
}

impl Drop for DirLock {
    fn drop(&mut self) {
        // A poisoned registry must not wedge every later open, and the record lock goes with
        // the descriptor either way.
        match held().lock() {
            Ok(mut held) => {
                held.remove(&self.key);
            }
            Err(poisoned) => {
                poisoned.into_inner().remove(&self.key);
            }
        }
    }
}

/// A successful acquisition.
pub(crate) struct Acquired {
    pub(crate) lock: DirLock,
    /// Set when the kernel reported that this filesystem does not implement record locking
    /// (some network mounts). Cross-process protection is then absent and the caller should
    /// say so; the in-process half still holds.
    pub(crate) unsupported: Option<io::Error>,
}

/// Takes `dir`'s write lock, or reports who holds it.
pub(crate) fn acquire(dir: &Path) -> Result<Acquired, LockFailure> {
    let path = dir.join(LOCK_FILE);
    // Key the registry on the canonical directory, so two spellings of one directory collide.
    // The lock file may not exist yet, hence canonicalizing the directory rather than it.
    let key = fs::canonicalize(dir).unwrap_or_else(|_| dir.to_path_buf());

    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&path)
        .map_err(|source| LockFailure::Io {
            path: path.clone(),
            source,
        })?;

    // In-process half first: a record lock would not refuse a second handle here.
    if !register(&key) {
        return Err(LockFailure::Contended {
            path,
            holder: Some(process::id()),
        });
    }
    let lock = DirLock {
        _file: file,
        key: key.clone(),
    };

    match take_record_lock(&lock._file) {
        Ok(()) => {
            record_holder(&lock._file);
            Ok(Acquired {
                lock,
                unsupported: None,
            })
        }
        Err(RecordLock::Contended) => {
            let holder = read_holder(&lock._file);
            drop(lock);
            Err(LockFailure::Contended { path, holder })
        }
        // Keep the lock: the registry is the only protection left, and refusing to open would
        // make tephra unusable on such a mount.
        Err(RecordLock::Unsupported(source)) => Ok(Acquired {
            lock,
            unsupported: Some(source),
        }),
        Err(RecordLock::Io(source)) => {
            drop(lock);
            Err(LockFailure::Io { path, source })
        }
    }
}

/// Adds `key` to the registry, returning whether it was not already there.
fn register(key: &Path) -> bool {
    let mut held = match held().lock() {
        Ok(held) => held,
        Err(poisoned) => poisoned.into_inner(),
    };
    held.insert(key.to_path_buf())
}

/// Why a record lock was not taken.
enum RecordLock {
    /// Another process holds it.
    Contended,
    /// The filesystem does not implement record locking.
    Unsupported(io::Error),
    /// Anything else: a real failure, not a verdict.
    Io(io::Error),
}

/// Takes a whole-file write lock owned by this process.
#[cfg(unix)]
fn take_record_lock(file: &File) -> Result<(), RecordLock> {
    use nix::errno::Errno;
    use nix::fcntl::{FcntlArg, fcntl};
    use nix::libc;

    let lock = libc::flock {
        l_type: libc::F_WRLCK as libc::c_short,
        l_whence: libc::SEEK_SET as libc::c_short,
        l_start: 0,
        // Zero means "to end of file", so the whole file regardless of length.
        l_len: 0,
        l_pid: 0,
    };
    match fcntl(file, FcntlArg::F_SETLK(&lock)) {
        Ok(_) => Ok(()),
        Err(Errno::EACCES | Errno::EAGAIN) => Err(RecordLock::Contended),
        // Only these mean "this filesystem cannot lock". Everything else (EIO, ENOLCK from a
        // full kernel lock table, EINTR) is a failure to be surfaced, not a licence to run
        // unprotected.
        Err(err @ (Errno::EOPNOTSUPP | Errno::ENOSYS)) => {
            Err(RecordLock::Unsupported(io::Error::from(err)))
        }
        Err(err) => Err(RecordLock::Io(io::Error::from(err))),
    }
}

/// Fallback for platforms without `fcntl`. Windows has no `fork`, so the descriptor-held
/// lock `try_lock` provides carries no inheritance hazard there.
#[cfg(not(unix))]
fn take_record_lock(file: &File) -> Result<(), RecordLock> {
    use std::fs::TryLockError;

    match file.try_lock() {
        Ok(()) => Ok(()),
        Err(TryLockError::WouldBlock) => Err(RecordLock::Contended),
        Err(TryLockError::Error(source)) => Err(RecordLock::Unsupported(source)),
    }
}

/// Records this process's id, so a contended open can name a holder. Best effort: a missing
/// or stale hint is never load-bearing.
#[cfg(unix)]
fn record_holder(file: &File) {
    let _ = file.set_len(0);
    let _ = (&*file).write_all(format!("{}\n", process::id()).as_bytes());
}

#[cfg(not(unix))]
fn record_holder(_file: &File) {}

/// Reads the pid a holder recorded, for the error message only.
#[cfg(unix)]
fn read_holder(file: &File) -> Option<u32> {
    let mut buf = [0u8; 32];
    let read = file.read_at(&mut buf, 0).ok()?;
    str::from_utf8(&buf[..read]).ok()?.trim().parse().ok()
}

#[cfg(not(unix))]
fn read_holder(_file: &File) -> Option<u32> {
    None
}