haematite 0.6.0

//! Binary persistence primitives shared by branch metadata stores.
//!
//! Branch metadata (the snapshot registry, the commit log, and — in later
//! briefs — branch handles) is small, append-mostly, and content-agnostic. This
//! module provides a length-prefixed little-endian codec ([`Reader`],
//! [`push_u64`], [`push_bytes`]) and an atomic file writer ([`write_atomic`])
//! that those stores build on, so each store only defines its own record shape.

use std::fmt;
use std::fs;
use std::io::Write;
use std::path::Path;

use crate::tree::Hash;
use crate::tree::node::HASH_SIZE;

const U64_LEN: usize = 8;

/// Error decoding or persisting a branch-metadata file.
#[derive(Debug)]
pub enum CodecError {
    /// The persisted bytes could not be decoded.
    Corrupt(String),
    /// An I/O error occurred while persisting.
    Io(std::io::Error),
}

impl fmt::Display for CodecError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Corrupt(reason) => write!(f, "branch metadata corrupted: {reason}"),
            Self::Io(error) => write!(f, "branch metadata I/O error: {error}"),
        }
    }
}

impl std::error::Error for CodecError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Corrupt(_) => None,
        }
    }
}

/// Appends `value` to `buffer` as little-endian bytes.
pub fn push_u64(buffer: &mut Vec<u8>, value: u64) {
    buffer.extend_from_slice(&value.to_le_bytes());
}

/// Appends `bytes` to `buffer`, prefixed by its length.
pub fn push_bytes(buffer: &mut Vec<u8>, bytes: &[u8]) {
    push_u64(buffer, bytes.len() as u64);
    buffer.extend_from_slice(bytes);
}

/// A forward-only reader over a persisted metadata file.
pub struct Reader<'a> {
    bytes: &'a [u8],
    offset: usize,
}

impl<'a> Reader<'a> {
    /// Wraps `bytes` for sequential decoding.
    pub const fn new(bytes: &'a [u8]) -> Self {
        Self { bytes, offset: 0 }
    }

    /// Consumes and verifies a fixed file-header magic.
    pub fn expect_magic(&mut self, magic: [u8; 4]) -> Result<(), CodecError> {
        let found = self.read_exact(magic.len())?;
        if found == magic.as_slice() {
            Ok(())
        } else {
            Err(CodecError::Corrupt("unrecognised file header".into()))
        }
    }

    /// Reads a little-endian `u64`.
    pub fn read_u64(&mut self) -> Result<u64, CodecError> {
        let bytes = self.read_exact(U64_LEN)?;
        let array: [u8; U64_LEN] = bytes
            .try_into()
            .map_err(|_error| CodecError::Corrupt("truncated integer".into()))?;
        Ok(u64::from_le_bytes(array))
    }

    /// Reads a length stored as a `u64`, narrowing to `usize`.
    pub fn read_usize(&mut self) -> Result<usize, CodecError> {
        let value = self.read_u64()?;
        usize::try_from(value)
            .map_err(|_error| CodecError::Corrupt("length exceeds platform width".into()))
    }

    /// Bytes not yet consumed — the honest ceiling for any count-derived
    /// pre-allocation (PERSIST-003 discipline: never allocate from an
    /// untrusted on-disk length; a corrupt count must fail as `Corrupt`,
    /// never abort the process at the allocator).
    pub const fn remaining(&self) -> usize {
        self.bytes.len().saturating_sub(self.offset)
    }

    /// Reads a length-prefixed byte string.
    pub fn read_bytes(&mut self) -> Result<Vec<u8>, CodecError> {
        let len = self.read_usize()?;
        Ok(self.read_exact(len)?.to_vec())
    }

    /// Reads a 32-byte content hash.
    pub fn read_hash(&mut self) -> Result<Hash, CodecError> {
        let bytes = self.read_exact(HASH_SIZE)?;
        let array: [u8; HASH_SIZE] = bytes
            .try_into()
            .map_err(|_error| CodecError::Corrupt("truncated hash".into()))?;
        Ok(Hash::from_bytes(array))
    }

    /// Confirms the whole input was consumed.
    pub fn finish(&self) -> Result<(), CodecError> {
        if self.offset == self.bytes.len() {
            Ok(())
        } else {
            Err(CodecError::Corrupt("trailing bytes after entries".into()))
        }
    }

    fn read_exact(&mut self, len: usize) -> Result<&'a [u8], CodecError> {
        let end = self
            .offset
            .checked_add(len)
            .ok_or_else(|| CodecError::Corrupt("length overflow".into()))?;
        let slice = self
            .bytes
            .get(self.offset..end)
            .ok_or_else(|| CodecError::Corrupt("unexpected end of file".into()))?;
        self.offset = end;
        Ok(slice)
    }
}

/// Outcome of a no-clobber atomic install ([`write_atomic_noclobber`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NoclobberOutcome {
    /// The bytes were durably installed at the target path.
    Installed,
    /// The target path already existed; nothing was written there. The caller
    /// decides whether that is a duplicate, a collision, or fine.
    TargetExists,
}

/// Failure of an atomic install ([`write_atomic`] / [`write_atomic_noclobber`]),
/// split by whether the rename landed before the failure.
///
/// The split is load-bearing for callers that mirror the target file in
/// memory (the branch ref store): after [`InstallError::InstalledUnfenced`]
/// the target path already holds the new bytes — a cold reopen, a crash
/// recovery, or any direct reader will see them — so a mirror that kept the
/// OLD contents would diverge from disk truth. Such callers must match on the
/// variant BEFORE converting; `From<InstallError> for CodecError` collapses
/// the distinction for callers with no mirror to keep.
#[derive(Debug)]
pub enum InstallError {
    /// The failure preceded the rename: the target path is untouched.
    NotInstalled(CodecError),
    /// The rename landed but the parent-directory fsync failed: the target
    /// path holds the new bytes, with unconfirmed directory-entry durability
    /// (power loss may still roll the entry back; process death will not).
    InstalledUnfenced(CodecError),
}

impl fmt::Display for InstallError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NotInstalled(error) => {
                write!(f, "atomic install failed before the rename: {error}")
            }
            Self::InstalledUnfenced(error) => write!(
                f,
                "atomic install renamed the target but the directory-entry fsync failed: {error}"
            ),
        }
    }
}

impl std::error::Error for InstallError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::NotInstalled(error) | Self::InstalledUnfenced(error) => Some(error),
        }
    }
}

impl From<InstallError> for CodecError {
    fn from(error: InstallError) -> Self {
        match error {
            InstallError::NotInstalled(inner) | InstallError::InstalledUnfenced(inner) => inner,
        }
    }
}

/// Atomically writes `bytes` to `path` via a temp file in the same directory.
///
/// A post-rename failure is reported as
/// [`InstallError::InstalledUnfenced`]: the target already holds the new
/// bytes and callers mirroring the file in memory must adopt them.
pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), InstallError> {
    write_atomic_unfenced(path, bytes).map_err(InstallError::NotInstalled)?;
    let parent = path.parent().ok_or_else(|| {
        InstallError::InstalledUnfenced(CodecError::Corrupt(
            "metadata path has no parent directory".into(),
        ))
    })?;
    sync_parent_dir(parent).map_err(InstallError::InstalledUnfenced)
}

/// Performs the byte-synced atomic replacement without its entry fence.
/// Durable-record adapters call [`sync_parent_dir`] as the seam's distinct
/// fifth operation immediately after this returns.
pub(crate) fn write_atomic_unfenced(path: &Path, bytes: &[u8]) -> Result<(), CodecError> {
    let (temp_file, _parent) = synced_temp_file(path, bytes)?;
    temp_file
        .persist(path)
        .map(drop)
        .map_err(|error| CodecError::Io(error.error))
}

/// Atomically writes `bytes` to `path`, refusing an existing target.
///
/// If `path` already exists nothing is installed and
/// [`NoclobberOutcome::TargetExists`] is returned. Reservation and creation
/// are one indivisible step: two racing creators can never both succeed, and
/// a crash leaves either the old world or the fully-synced new file, never a
/// torn or half-claimed name.
///
/// Platform honesty (mirrors the [`sync_parent_dir`] caveat): on Linux/macOS
/// the no-replace install is `renameat2(..., RENAME_NOREPLACE)` with a
/// `link(2)`+`unlink` fallback — both refuse an existing target atomically at
/// the kernel. On Windows `tempfile` uses `MoveFileExW` *without*
/// `MOVEFILE_REPLACE_EXISTING`, which also refuses an existing target, but the
/// directory-entry fsync is skipped there (see [`sync_parent_dir`]), so the
/// install's crash-durability — not its exclusivity — is weaker on Windows.
///
/// As with [`write_atomic`], a post-rename failure is
/// [`InstallError::InstalledUnfenced`]: the target now exists and holds the
/// new bytes.
pub fn write_atomic_noclobber(path: &Path, bytes: &[u8]) -> Result<NoclobberOutcome, InstallError> {
    let outcome =
        write_atomic_noclobber_unfenced(path, bytes).map_err(InstallError::NotInstalled)?;
    if outcome == NoclobberOutcome::Installed {
        let parent = path.parent().ok_or_else(|| {
            InstallError::InstalledUnfenced(CodecError::Corrupt(
                "metadata path has no parent directory".into(),
            ))
        })?;
        sync_parent_dir(parent).map_err(InstallError::InstalledUnfenced)?;
    }
    Ok(outcome)
}

/// Performs the byte-synced no-clobber install without its entry fence.
pub(crate) fn write_atomic_noclobber_unfenced(
    path: &Path,
    bytes: &[u8],
) -> Result<NoclobberOutcome, CodecError> {
    let (temp_file, _parent) = synced_temp_file(path, bytes)?;
    match temp_file.persist_noclobber(path) {
        Ok(_persisted) => Ok(NoclobberOutcome::Installed),
        // The temp file rides back inside the error and is unlinked on drop,
        // so a lost race leaves no stray temp behind.
        Err(error) if error.error.kind() == std::io::ErrorKind::AlreadyExists => {
            Ok(NoclobberOutcome::TargetExists)
        }
        Err(error) => Err(CodecError::Io(error.error)),
    }
}

/// Writes `bytes` to a fresh temp file beside `path` and syncs its contents,
/// ready for either the clobbering or the no-clobber rename install.
///
/// The temp prefix/suffix are pinned to `.branch-`/`.tmp` (the same convention
/// as `db/config.rs`'s `.config-`): [`crate::branch::refstore::BranchRefStore::open`]
/// sweeps orphaned temp files by exactly this pattern, so every temp file this
/// module creates MUST match it or a crash mid-install leaks an unsweepable file.
fn synced_temp_file<'path>(
    path: &'path Path,
    bytes: &[u8],
) -> Result<(tempfile::NamedTempFile, &'path Path), CodecError> {
    let parent = path
        .parent()
        .ok_or_else(|| CodecError::Corrupt("metadata path has no parent directory".into()))?;
    fs::create_dir_all(parent).map_err(CodecError::Io)?;

    let mut temp_file = tempfile::Builder::new()
        .prefix(".branch-")
        .suffix(".tmp")
        .tempfile_in(parent)
        .map_err(CodecError::Io)?;
    temp_file.write_all(bytes).map_err(CodecError::Io)?;
    temp_file.as_file_mut().sync_all().map_err(CodecError::Io)?;
    Ok((temp_file, parent))
}

/// Flushes the directory entry created by the atomic rename.
///
/// The rename is durable only once the directory entry itself is synced: a
/// synced file with an unsynced directory entry can still vanish on power loss,
/// leaving the metadata apparently empty on restart.
///
/// This is Unix-only. Opening a directory to fsync it requires
/// `FILE_FLAG_BACKUP_SEMANTICS` on Windows, which `std` does not set — there
/// `File::open` on a directory hard-errors rather than no-ops, so attempting it
/// would fail every persist. On non-Unix platforms the directory-entry sync is
/// therefore skipped; the file contents are still synced via `sync_all` above,
/// but crash-durability of the directory entry is not guaranteed there.
#[cfg(unix)]
pub(crate) fn sync_parent_dir(parent: &Path) -> Result<(), CodecError> {
    #[cfg(test)]
    if take_injected_parent_dir_sync_failure() {
        return Err(CodecError::Io(std::io::Error::other(
            "injected parent-directory fsync failure",
        )));
    }
    fs::File::open(parent)
        .and_then(|dir| dir.sync_all())
        .map_err(CodecError::Io)
}

// Test-only seam mirroring `tree::mutate`'s: lets the crash battery drive the
// REAL install code path into its post-rename fsync-failure tail (the
// `InstallError::InstalledUnfenced` arm), which no real filesystem can be
// asked to produce on demand. Compiled out entirely in non-test builds.
#[cfg(test)]
thread_local! {
    static FAIL_NEXT_PARENT_DIR_SYNC: std::cell::Cell<bool> =
        const { std::cell::Cell::new(false) };
}

/// Arms the failpoint: the NEXT [`sync_parent_dir`] call on this thread fails
/// with an injected I/O error (then the failpoint disarms itself).
#[cfg(test)]
pub(crate) fn fail_next_parent_dir_sync() {
    FAIL_NEXT_PARENT_DIR_SYNC.with(|flag| flag.set(true));
}

#[cfg(test)]
fn take_injected_parent_dir_sync_failure() -> bool {
    FAIL_NEXT_PARENT_DIR_SYNC.with(std::cell::Cell::take)
}

#[cfg(not(unix))]
pub(crate) fn sync_parent_dir(_parent: &Path) -> Result<(), CodecError> {
    Ok(())
}