haematite 0.7.0

//! BRANCH-003: Snapshot registry and commit log.
//!
//! The snapshot registry (R1) maps human-readable names to committed root
//! hashes and persists across restarts. The commit log (R2) records every
//! committed root hash in commit order with a timestamp. Snapshot listing (R5)
//! returns named snapshots with their hashes and timestamps in chronological
//! order.
//!
//! Timestamp semantics: an entry's timestamp is *naming time* (when `name` was
//! called), not the root's commit time (which lives in the [`CommitLog`]).

use std::collections::HashMap;
use std::fmt;
use std::fs;
use std::path::{Path, PathBuf};

use super::persist::{CodecError, Reader, push_bytes, push_u64, write_atomic};
use crate::tree::Hash;

pub use super::time::{Timestamp, current_timestamp};

const REGISTRY_MAGIC: &[u8; 4] = b"HSR1";
const LOG_MAGIC: &[u8; 4] = b"HCL1";

/// One named snapshot: a name bound to a committed root hash at a point in time.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SnapshotEntry {
    pub name: String,
    pub root_hash: Hash,
    pub timestamp: Timestamp,
}

/// One commit-log record: a committed root hash and when it was committed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommitLogEntry {
    pub root_hash: Hash,
    pub timestamp: Timestamp,
}

/// Errors raised by the snapshot registry and commit log.
#[derive(Debug)]
pub enum SnapshotError {
    /// A snapshot with this name already exists.
    DuplicateName(String),
    /// The persisted file could not be decoded.
    Corrupt(String),
    /// An I/O error occurred while persisting or loading.
    Io(std::io::Error),
}

impl fmt::Display for SnapshotError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::DuplicateName(name) => write!(f, "snapshot name already exists: {name}"),
            Self::Corrupt(reason) => write!(f, "snapshot store corrupted: {reason}"),
            Self::Io(error) => write!(f, "snapshot store I/O error: {error}"),
        }
    }
}

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

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

impl From<CodecError> for SnapshotError {
    fn from(error: CodecError) -> Self {
        match error {
            CodecError::Corrupt(reason) => Self::Corrupt(reason),
            CodecError::Io(io_error) => Self::Io(io_error),
        }
    }
}

/// Persistent mapping from human-readable names to committed root hashes.
///
/// Names are unique: binding a name that already exists is an error. Entries
/// retain insertion order so that [`SnapshotRegistry::list_snapshots`] reports
/// them chronologically. When constructed with [`SnapshotRegistry::open`], every
/// mutation is flushed to disk so the registry survives a restart.
#[derive(Debug)]
pub struct SnapshotRegistry {
    entries: Vec<SnapshotEntry>,
    index: HashMap<String, Hash>,
    path: Option<PathBuf>,
}

impl SnapshotRegistry {
    /// Creates an empty, in-memory registry with no backing file.
    pub fn new() -> Self {
        Self {
            entries: Vec::new(),
            index: HashMap::new(),
            path: None,
        }
    }

    /// Opens (or creates) a registry persisted at `path`, loading any entries.
    ///
    /// FENCE-CHAIN r8 R1d/L4: the containing directory is ESTABLISHED and its own
    /// directory entry FENCED here (D1 moved the creation off the first write —
    /// `synced_temp_file` no longer `create_dir_all`s). The metadata file's entry
    /// inside the containing dir is fenced later by [`write_atomic`].
    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, SnapshotError> {
        let path = path.as_ref().to_path_buf();
        establish_containing_dir(&path)?;
        let entries = match fs::read(&path) {
            Ok(bytes) => decode_registry(&bytes)?,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
            Err(error) => return Err(SnapshotError::Io(error)),
        };
        let mut index = HashMap::with_capacity(entries.len());
        for entry in &entries {
            index.insert(entry.name.clone(), entry.root_hash);
        }
        Ok(Self {
            entries,
            index,
            path: Some(path),
        })
    }

    /// Binds `name` to `root_hash`, stamped with the current naming time.
    ///
    /// Returns [`SnapshotError::DuplicateName`] if the name is already taken,
    /// leaving the registry unchanged.
    pub fn name(&mut self, name: &str, root_hash: Hash) -> Result<(), SnapshotError> {
        self.name_at(name, root_hash, current_timestamp())
    }

    /// Binds `name` to `root_hash` with an explicit timestamp.
    pub fn name_at(
        &mut self,
        name: &str,
        root_hash: Hash,
        timestamp: Timestamp,
    ) -> Result<(), SnapshotError> {
        if self.index.contains_key(name) {
            return Err(SnapshotError::DuplicateName(name.to_owned()));
        }
        self.entries.push(SnapshotEntry {
            name: name.to_owned(),
            root_hash,
            timestamp,
        });
        self.index.insert(name.to_owned(), root_hash);
        if let Err(error) = self.persist() {
            // Roll the in-memory state back so it stays consistent with disk.
            self.entries.pop();
            self.index.remove(name);
            return Err(error);
        }
        Ok(())
    }

    /// Returns the root hash bound to `name`, or `None` if unknown.
    pub fn get(&self, name: &str) -> Option<Hash> {
        self.index.get(name).copied()
    }

    /// Removes the snapshot bound to `name`, returning the removed entry.
    ///
    /// Unknown names return `Ok(None)`. Persisted registries flush the removal
    /// atomically; if persistence fails, the in-memory entry and index are
    /// restored so memory remains consistent with disk.
    pub fn remove(&mut self, name: &str) -> Result<Option<SnapshotEntry>, SnapshotError> {
        let Some(position) = self.entries.iter().position(|entry| entry.name == name) else {
            return Ok(None);
        };

        let removed = self.entries.remove(position);
        self.index.remove(name);

        if let Err(error) = self.persist() {
            self.entries.insert(position, removed.clone());
            self.index.insert(removed.name.clone(), removed.root_hash);
            return Err(error);
        }

        Ok(Some(removed))
    }

    /// Lists every named snapshot as `(name, root_hash, timestamp)` tuples in
    /// chronological (naming) order; `timestamp` is naming time (see module doc).
    pub fn list_snapshots(&self) -> Vec<(String, Hash, Timestamp)> {
        self.entries
            .iter()
            .map(|entry| (entry.name.clone(), entry.root_hash, entry.timestamp))
            .collect()
    }

    /// Number of named snapshots.
    pub const fn len(&self) -> usize {
        self.entries.len()
    }

    /// Whether the registry holds no snapshots.
    pub const fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    fn persist(&self) -> Result<(), SnapshotError> {
        if let Some(path) = &self.path {
            // No mirror of the FILE is kept beyond `entries`, which both
            // install outcomes leave authoritative — the unfenced/not-
            // installed split write_atomic reports is collapsed here.
            write_atomic(path, &encode_registry(&self.entries)).map_err(CodecError::from)?;
        }
        Ok(())
    }
}

impl Default for SnapshotRegistry {
    fn default() -> Self {
        Self::new()
    }
}

/// Append-only log of every committed root hash, in commit order.
///
/// `Database::commit` appends each new composite root hash here together with a
/// timestamp. When constructed with [`CommitLog::open`], every append is flushed
/// to disk so the log survives a restart.
#[derive(Debug)]
pub struct CommitLog {
    entries: Vec<CommitLogEntry>,
    path: Option<PathBuf>,
}

impl CommitLog {
    /// Creates an empty, in-memory log with no backing file.
    pub const fn new() -> Self {
        Self {
            entries: Vec::new(),
            path: None,
        }
    }

    /// Opens (or creates) a log persisted at `path`, loading any existing entries.
    ///
    /// FENCE-CHAIN r8 R1d/L4: establishes and fences the containing directory's
    /// own entry, exactly as [`SnapshotRegistry::open`] does.
    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, SnapshotError> {
        let path = path.as_ref().to_path_buf();
        establish_containing_dir(&path)?;
        let entries = match fs::read(&path) {
            Ok(bytes) => decode_log(&bytes)?,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
            Err(error) => return Err(SnapshotError::Io(error)),
        };
        Ok(Self {
            entries,
            path: Some(path),
        })
    }

    /// Appends `root_hash` to the log with the given commit `timestamp`.
    pub fn append(&mut self, root_hash: Hash, timestamp: Timestamp) -> Result<(), SnapshotError> {
        self.entries.push(CommitLogEntry {
            root_hash,
            timestamp,
        });
        if let Err(error) = self.persist() {
            self.entries.pop();
            return Err(error);
        }
        Ok(())
    }

    /// Lists every commit-log entry in chronological (commit) order.
    pub const fn list(&self) -> &[CommitLogEntry] {
        self.entries.as_slice()
    }

    /// Number of recorded commits.
    pub const fn len(&self) -> usize {
        self.entries.len()
    }

    /// Whether the log holds no commits.
    pub const fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    fn persist(&self) -> Result<(), SnapshotError> {
        if let Some(path) = &self.path {
            // Same collapse as the registry above: `entries` stays
            // authoritative on every install outcome.
            write_atomic(path, &encode_log(&self.entries)).map_err(CodecError::from)?;
        }
        Ok(())
    }
}

impl Default for CommitLog {
    fn default() -> Self {
        Self::new()
    }
}

/// Establish and fence the containing directory of a metadata file (L4/R1d/D1).
///
/// D1: the containing directory is created a SINGLE level below its (pre-existing)
/// parent, accepting an existing directory (D2) and a concurrent creator's win.
/// L4: the containing directory's OWN entry — which lives in ITS parent — is
/// fenced UNCONDITIONALLY on every open (per-open fence, causal on retry), closing
/// the gap where a lazily-created containing directory rode unfenced past a durable
/// metadata write. In test builds the creation is journalled so a CUT can rewind
/// an unfenced containing directory.
fn establish_containing_dir(path: &Path) -> Result<(), SnapshotError> {
    let Some(containing) = path.parent() else {
        return Ok(());
    };
    let containing = if containing.as_os_str().is_empty() {
        Path::new(".")
    } else {
        containing
    };
    match fs::metadata(containing) {
        Ok(metadata) if metadata.is_dir() => {}
        Ok(_metadata) => {
            return Err(SnapshotError::Io(std::io::Error::other(format!(
                "snapshot containing path is not a directory: {}",
                containing.display()
            ))));
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            #[cfg(test)]
            let reservation = crate::fence::journal::reserve_create_dir(containing);
            match fs::create_dir(containing) {
                Ok(()) => {
                    #[cfg(test)]
                    reservation.commit();
                }
                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                    #[cfg(test)]
                    reservation.cancel();
                }
                Err(error) => {
                    #[cfg(test)]
                    reservation.cancel();
                    return Err(SnapshotError::Io(error));
                }
            }
        }
        Err(error) => return Err(SnapshotError::Io(error)),
    }

    if let Some(grandparent) = containing.parent() {
        let grandparent = if grandparent.as_os_str().is_empty() {
            Path::new(".")
        } else {
            grandparent
        };
        crate::fence::sync_dir_entry(grandparent).map_err(SnapshotError::Io)?;
    }
    Ok(())
}

fn encode_registry(entries: &[SnapshotEntry]) -> Vec<u8> {
    let mut bytes = Vec::new();
    bytes.extend_from_slice(REGISTRY_MAGIC);
    push_u64(&mut bytes, entries.len() as u64);
    for entry in entries {
        push_bytes(&mut bytes, entry.name.as_bytes());
        bytes.extend_from_slice(entry.root_hash.as_bytes());
        push_u64(&mut bytes, entry.timestamp);
    }
    bytes
}

/// Pure HSR1 registry decode, shared BY CONSTRUCTION with the vacuum's
/// read-only snapshot reader (STORAGE-VACUUM.md §7 ⟨r4, M4⟩): one parser, two
/// callers — [`SnapshotRegistry::open`] keeps calling exactly this function,
/// and the vacuum cannot accept or reject a registry the production store
/// wouldn't.
pub(crate) fn decode_registry(bytes: &[u8]) -> Result<Vec<SnapshotEntry>, SnapshotError> {
    let mut reader = Reader::new(bytes);
    reader.expect_magic(*REGISTRY_MAGIC)?;
    let count = reader.read_usize()?;
    // Capacity is a HINT capped by what the remaining bytes could possibly
    // hold (each entry is ≥ 48 bytes serialized): a corrupt count fails as
    // Corrupt when the bytes run out, never aborts at the allocator.
    let mut entries = Vec::with_capacity(count.min(reader.remaining() / 48));
    for _ in 0..count {
        let name_bytes = reader.read_bytes()?;
        let name = String::from_utf8(name_bytes)
            .map_err(|_error| SnapshotError::Corrupt("snapshot name is not valid UTF-8".into()))?;
        let root_hash = reader.read_hash()?;
        let timestamp = reader.read_u64()?;
        entries.push(SnapshotEntry {
            name,
            root_hash,
            timestamp,
        });
    }
    reader.finish()?;
    Ok(entries)
}

fn encode_log(entries: &[CommitLogEntry]) -> Vec<u8> {
    let mut bytes = Vec::new();
    bytes.extend_from_slice(LOG_MAGIC);
    push_u64(&mut bytes, entries.len() as u64);
    for entry in entries {
        bytes.extend_from_slice(entry.root_hash.as_bytes());
        push_u64(&mut bytes, entry.timestamp);
    }
    bytes
}

fn decode_log(bytes: &[u8]) -> Result<Vec<CommitLogEntry>, SnapshotError> {
    let mut reader = Reader::new(bytes);
    reader.expect_magic(*LOG_MAGIC)?;
    let count = reader.read_usize()?;
    // Same PERSIST-003 discipline as decode_registry: each entry is exactly
    // 40 serialized bytes, so the hint never exceeds the input.
    let mut entries = Vec::with_capacity(count.min(reader.remaining() / 40));
    for _ in 0..count {
        let root_hash = reader.read_hash()?;
        let timestamp = reader.read_u64()?;
        entries.push(CommitLogEntry {
            root_hash,
            timestamp,
        });
    }
    reader.finish()?;
    Ok(entries)
}

#[cfg(test)]
#[path = "snapshot_tests.rs"]
mod tests;