haematite 0.7.0

//! Filesystem adapter for the native-neutral durable-record seam.

use std::fs;
use std::path::{Path, PathBuf};

use super::durable_record::{CreateExclusive, DurableRecordStore, EntryFence};
use super::persist::{
    NoclobberOutcome, sync_parent_dir, write_atomic_noclobber_unfenced, write_atomic_unfenced,
};
use super::refstore::codec::{corrupt_in_file, encode_record};
use super::refstore::{
    BranchRefError, BranchRefRecord, REF_EXTENSION, TEMP_PREFIX, TEMP_SUFFIX, decode_record,
    ref_file_name,
};
use super::time::Timestamp;

/// Existing native one-file-per-HBR1 adapter.
#[derive(Debug)]
pub(super) struct NativeDurableRecordStore {
    dir: PathBuf,
}

impl NativeDurableRecordStore {
    pub(super) fn open<P: AsRef<Path>>(dir: P) -> Result<Self, BranchRefError> {
        let dir = dir.as_ref().to_path_buf();
        // D1: single-level create below the pre-existing `data_dir`, accept
        // existing (D2). A missing parent surfaces as the natural `NotFound`.
        Self::ensure_refs_dir_impl(&dir)?;
        // D2/R1d: fence the refs-dir entry UNCONDITIONALLY on every open — a
        // per-open fence (R2), causal on cold retry (R4d), so a refs dir whose
        // entry never reached the platter is re-fenced by the next open.
        if let Some(parent) = dir.parent() {
            let parent = if parent.as_os_str().is_empty() {
                Path::new(".")
            } else {
                parent
            };
            sync_parent_dir(parent)?;
        }
        Ok(Self { dir })
    }

    fn ref_path(&self, name: &str) -> PathBuf {
        self.dir.join(ref_file_name(name))
    }

    /// Establish the refs directory a single level below its pre-existing parent
    /// (D1), accepting an existing directory (D2) and a concurrent creator's win.
    /// In test builds the creation is journalled so a CUT can rewind an unfenced
    /// refs-dir entry.
    fn ensure_refs_dir_impl(dir: &Path) -> Result<(), BranchRefError> {
        match fs::metadata(dir) {
            Ok(metadata) if metadata.is_dir() => Ok(()),
            Ok(_metadata) => Err(BranchRefError::Corrupt(format!(
                "ref store path is not a directory: {}",
                dir.display()
            ))),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                #[cfg(test)]
                let reservation = crate::fence::journal::reserve_create_dir(dir);
                match fs::create_dir(dir) {
                    Ok(()) => {
                        #[cfg(test)]
                        reservation.commit();
                        Ok(())
                    }
                    Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                        #[cfg(test)]
                        reservation.cancel();
                        Ok(())
                    }
                    Err(error) => {
                        #[cfg(test)]
                        reservation.cancel();
                        Err(BranchRefError::Io(error))
                    }
                }
            }
            Err(error) => Err(BranchRefError::Io(error)),
        }
    }

    fn read_record(path: &Path) -> Result<BranchRefRecord, BranchRefError> {
        decode_record(&fs::read(path)?).map_err(|error| corrupt_in_file(path, &error))
    }
}

impl DurableRecordStore for NativeDurableRecordStore {
    type Error = BranchRefError;
    fn list_read_at_open(&mut self) -> Result<Vec<BranchRefRecord>, BranchRefError> {
        let mut records = Vec::new();
        for entry in fs::read_dir(&self.dir)? {
            let entry = entry?;
            let path = entry.path();
            let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
                continue;
            };
            if file_name.starts_with(TEMP_PREFIX) && file_name.ends_with(TEMP_SUFFIX) {
                fs::remove_file(&path)?;
                continue;
            }
            if path.extension().and_then(|extension| extension.to_str()) != Some(REF_EXTENSION) {
                continue;
            }
            let record = Self::read_record(&path)?;
            if ref_file_name(&record.name) != file_name {
                return Err(BranchRefError::Corrupt(format!(
                    "ref file {} holds a record for branch {:?}, whose name hashes elsewhere",
                    path.display(),
                    record.name,
                )));
            }
            records.push(record);
        }
        Ok(records)
    }

    fn create_exclusive(
        &mut self,
        record: &BranchRefRecord,
    ) -> Result<CreateExclusive, BranchRefError> {
        let path = self.ref_path(&record.name);
        match write_atomic_noclobber_unfenced(&path, &encode_record(record))? {
            NoclobberOutcome::Installed => Ok(CreateExclusive::Installed),
            NoclobberOutcome::TargetExists => {
                Ok(CreateExclusive::TargetExists(Self::read_record(&path)?))
            }
        }
    }

    fn cas_replace_install(
        &mut self,
        name: &str,
        expected_created: Timestamp,
        expected_seq: u64,
        replacement: &BranchRefRecord,
    ) -> Result<BranchRefRecord, BranchRefError> {
        let path = self.ref_path(name);
        let existing = match Self::read_record(&path) {
            Ok(record) => record,
            Err(BranchRefError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {
                return Err(BranchRefError::BranchRemoved(name.to_owned()));
            }
            Err(error) => return Err(error),
        };
        if existing.created != expected_created {
            return Err(BranchRefError::BranchGenerationMismatch {
                name: name.to_owned(),
                expected_created,
                found_created: existing.created,
            });
        }
        if existing.seq != expected_seq {
            return Err(BranchRefError::StaleSeq {
                name: name.to_owned(),
                expected: expected_seq,
                found: existing.seq,
            });
        }
        write_atomic_unfenced(&path, &encode_record(replacement))?;
        Ok(replacement.clone())
    }

    fn unlink(&mut self, name: &str) -> Result<bool, BranchRefError> {
        match fs::remove_file(self.ref_path(name)) {
            Ok(()) => Ok(true),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(error) => Err(BranchRefError::Io(error)),
        }
    }

    fn entry_fence(&mut self, _expected: EntryFence<'_>) -> Result<(), BranchRefError> {
        sync_parent_dir(&self.dir).map_err(BranchRefError::from)
    }
}