haematite 0.7.0

//! The on-disk content-addressed node store type.

use std::cell::RefCell;
use std::collections::BTreeSet;
use std::fs;
use std::io::ErrorKind;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use crate::store::NodeStore;
use crate::store::cache::LruCache;
use crate::store::gc::DeleteNode;
use crate::tree::{Hash, Node};

use super::barrier::{drain_and_fence, ensure_prefix_dir};
use super::construct::{ConstructMode, ensure_directory};
use super::error::StoreError;
use super::node_io::{compress_node, decompress_node, path_exists, publish_compressed_node};

const DEFAULT_CACHE_CAPACITY: usize = 1_024;
const HASH_PREFIX_HEX_LEN: usize = 2;

#[derive(Debug)]
pub struct DiskStore {
    dir: PathBuf,
    cache: RefCell<LruCache>,
    /// Parent subdirectories that received a node file since the last
    /// [`Self::sync_dirty_dirs`] barrier (Tier-0 durability fix).
    ///
    /// `write_compressed_node` syncs each node file's DATA but does NOT fsync the
    /// parent directory, so a published rename can be lost on power loss even
    /// though the bytes are durable. Each distinct parent dir is recorded here
    /// (deduped — a commit writes many nodes into few subdirs) and fsync'd ONCE
    /// as a batched barrier by [`Self::sync_dirty_dirs`], which the commit path
    /// invokes strictly before the WAL committed-root marker is written.
    dirty_dirs: RefCell<BTreeSet<PathBuf>>,
    /// FENCE-CHAIN r8 R1c (L3): directories owed a fence to make a newly-created
    /// CHILD DIRECTORY entry durable — the store root, joined the moment a new
    /// prefix subdirectory is created, so the barrier fsyncs the store root and
    /// the new subdir's own directory entry survives power loss (the node files'
    /// entries live one level deeper and are fenced via `dirty_dirs`). Kept in a
    /// sibling set so the node-file barrier's set and tests are unchanged, while
    /// the same [`Self::sync_dirty_dirs`] discharges both obligations before the
    /// marker. Retention on a failed barrier holds it here until its fsync lands.
    dirty_entry_fences: RefCell<BTreeSet<PathBuf>>,
}

impl DiskStore {
    /// Open (creating if needed) a node store rooted at `dir`.
    ///
    /// OWNERSHIP RULE (STORAGE-VACUUM.md §6, normative): directories under a
    /// database `data_dir` are owned by that database. Opening a `DiskStore`
    /// directly on `<data_dir>/shard-{id}/store` bypasses the A4 data-dir
    /// writer lock entirely — the lock excludes only writers routed through a
    /// live `Database` — and such direct access FORFEITS vacuum safety: an
    /// offline vacuum holding the lock cannot see this writer, and its sweep
    /// may delete nodes the bypassing writer just published. Point standalone
    /// stores at directories no database owns.
    pub fn new<P>(dir: P) -> Result<Self, StoreError>
    where
        P: AsRef<Path>,
    {
        Self::with_cache_capacity(dir, DEFAULT_CACHE_CAPACITY)
    }

    /// Open with an explicit LRU cache capacity. The ownership rule on
    /// [`Self::new`] applies identically.
    ///
    /// FENCE-CHAIN r8 D3 (mode 1, `Create-or-open AND fence`): this is the public
    /// door BOTH constructors resolve to — it owns the `store` root's directory
    /// entry, so a fresh store's entry in its parent (`shard-{id}`) is fsynced
    /// here, closing L2. A fence failure is a synchronous [`StoreError::Io`]; no
    /// new failure mode reaches D3's crate-private callers.
    pub fn with_cache_capacity<P>(dir: P, cache_capacity: usize) -> Result<Self, StoreError>
    where
        P: AsRef<Path>,
    {
        Self::construct(
            dir.as_ref(),
            cache_capacity,
            ConstructMode::CreateOrOpenFenced,
        )
    }

    /// FENCE-CHAIN r8 D3 (mode 2, `Create-or-open UNFENCED`, crate-private): the
    /// native boot path (`shard::actor::native`) ONLY. A fresh shard still needs
    /// its `store` root CREATED before `DurableWal::new` constructs, but the
    /// entry's fence is owned transitively by the WAL's parent fsync of the
    /// SHARED shard directory (ordering pinned by R1e — `DiskStore` before
    /// `DurableWal`), so this door creates without fencing. No wall covers it;
    /// the fresh-shard native path is covered by the L1/R3 crash pin.
    pub(crate) fn new_unfenced<P>(dir: P) -> Result<Self, StoreError>
    where
        P: AsRef<Path>,
    {
        Self::construct(
            dir.as_ref(),
            DEFAULT_CACHE_CAPACITY,
            ConstructMode::CreateOrOpenUnfenced,
        )
    }

    /// FENCE-CHAIN r8 D3 (mode 3, `Open-existing UNFENCED`, crate-private): the
    /// read-only observer (`db::observer`) ONLY. It REFUSES to create — an
    /// observer that minted a store root would violate its never-mutates promise
    /// — and never fences (no entry it owns to make durable).
    pub(crate) fn open_existing<P>(dir: P) -> Result<Self, StoreError>
    where
        P: AsRef<Path>,
    {
        Self::construct(
            dir.as_ref(),
            DEFAULT_CACHE_CAPACITY,
            ConstructMode::OpenExistingUnfenced,
        )
    }

    fn construct(
        dir: &Path,
        cache_capacity: usize,
        mode: ConstructMode,
    ) -> Result<Self, StoreError> {
        let dir = dir.to_path_buf();
        let cache = LruCache::new(cache_capacity).map_err(StoreError::from)?;
        ensure_directory(&dir, mode)?;

        Ok(Self {
            dir,
            cache: RefCell::new(cache),
            dirty_dirs: RefCell::new(BTreeSet::new()),
            dirty_entry_fences: RefCell::new(BTreeSet::new()),
        })
    }

    pub fn cache_capacity(&self) -> usize {
        self.cache.borrow().capacity()
    }

    pub fn get(&self, hash: &Hash) -> Result<Option<Arc<Node>>, StoreError> {
        self.read_node(hash)
    }

    pub fn put(&mut self, node: &Node) -> Result<Hash, StoreError> {
        self.write_node(node)
    }

    pub fn delete(&self, hash: &Hash) -> Result<(), StoreError> {
        self.delete_node(hash)
    }

    fn read_node(&self, hash: &Hash) -> Result<Option<Arc<Node>>, StoreError> {
        if let Some(node) = self.cache_get(hash) {
            return Ok(Some(node));
        }

        let path = self.node_path(hash);
        let compressed = match fs::read(path) {
            Ok(bytes) => bytes,
            Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(StoreError::Io(error)),
        };
        let serialised = decompress_node(&compressed)?;
        let node = Arc::new(
            Node::deserialise(&serialised)
                .map_err(|error| StoreError::Deserialise(error.to_string()))?,
        );
        self.cache_put(*hash, Arc::clone(&node));
        Ok(Some(node))
    }

    fn write_node(&self, node: &Node) -> Result<Hash, StoreError> {
        // Serialise ONCE: `serialise_and_hash` yields the content hash and the
        // exact bytes we hand to compression, so a node is no longer serialised
        // twice per write (previously `hash()` serialised-then-dropped the bytes,
        // and the compress step serialised again).
        let (serialised, hash) = node.serialise_and_hash();
        let path = self.node_path(&hash);
        if path_exists(&path)? {
            self.cache_put(hash, Arc::new(node.clone()));
            return Ok(hash);
        }

        let compressed = compress_node(&serialised)?;

        let parent_dir = path
            .parent()
            .ok_or_else(|| StoreError::MissingParentDirectory { path: path.clone() })?;
        // R1c (L3, obligation-first): establish the node's prefix subdirectory,
        // and the MOMENT a NEW one is created retain the STORE ROOT for fencing
        // that subdir's directory entry — recorded BEFORE any fallible publish
        // step below, so a failed-then-retried publish still fences the entry.
        // Retention holds the store root in the dirty set until its fsync
        // succeeds; the race rule (a concurrent creator won) is the retained
        // obligation the barrier discharges.
        if ensure_prefix_dir(parent_dir)? {
            self.dirty_entry_fences
                .borrow_mut()
                .insert(self.dir.clone());
        }

        if let Some(persisted_parent) = publish_compressed_node(&path, &compressed)? {
            // Record the subdirectory that received this node so the commit-path
            // barrier (`sync_dirty_dirs`) can fsync its directory entry ONCE,
            // deduped, before the WAL marker is written. A `None` here means the
            // file already existed (no new directory entry to make durable).
            self.dirty_dirs.borrow_mut().insert(persisted_parent);
        }
        self.cache_put(hash, Arc::new(node.clone()));
        Ok(hash)
    }

    /// fsync the directory entry of every subdirectory that received a node
    /// since the last barrier, then clear the dirty set (Tier-0 durability fix).
    ///
    /// Each distinct directory is opened read-only and `sync_all`'d once — the
    /// portable Unix idiom for making a rename's directory entry durable. This
    /// is the batched barrier the commit path runs AFTER all of a commit's nodes
    /// are persisted and STRICTLY BEFORE the WAL committed-root marker is
    /// written.
    fn sync_dirty_directories(&self) -> Result<(), StoreError> {
        // A directory leaves the set only AFTER its fsync succeeds. Taking
        // the whole set up front and failing mid-way would silently forget
        // the un-fenced remainder — the documented "barrier failed, retry the
        // commit" path would then re-run against an empty set, and a durable
        // record could be installed naming roots whose directory entries were
        // never fenced (power loss ⇒ record points at vanished nodes).
        //
        // L3 first: the store-root entry fences (a new prefix subdir's OWN entry)
        // are discharged before the node subdirs. Both go through the one
        // fence-chain primitive so the store-root fence clears its journal
        // obligation and the OBSERVER counts every real directory fsync; node
        // subdirs keep the pre-existing batched-barrier semantics. Retention on a
        // failed fsync holds the un-fenced remainder in its own set.
        drain_and_fence(&self.dirty_entry_fences)?;
        drain_and_fence(&self.dirty_dirs)
    }

    fn delete_node(&self, hash: &Hash) -> Result<(), StoreError> {
        self.cache_remove(hash);
        match fs::remove_file(self.node_path(hash)) {
            Ok(()) => Ok(()),
            Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
            Err(error) => Err(StoreError::Io(error)),
        }
    }

    fn node_path(&self, hash: &Hash) -> PathBuf {
        let hex = hash.to_string();
        let (prefix, file_name) = hex.split_at(HASH_PREFIX_HEX_LEN);
        self.dir.join(prefix).join(file_name)
    }

    fn cache_get(&self, hash: &Hash) -> Option<Arc<Node>> {
        self.cache.borrow_mut().get(hash)
    }

    fn cache_put(&self, hash: Hash, node: Arc<Node>) {
        self.cache.borrow_mut().put(hash, node);
    }

    fn cache_remove(&self, hash: &Hash) -> Option<Arc<Node>> {
        self.cache.borrow_mut().remove(hash)
    }

    /// The distinct parent subdirectories awaiting a directory-entry fsync, in
    /// sorted order (test-support for the durability barrier).
    #[cfg(test)]
    fn pending_dirty_dirs(&self) -> Vec<PathBuf> {
        self.dirty_dirs.borrow().iter().cloned().collect()
    }

    /// The store roots awaiting an L3 entry fence (a new prefix subdir's own
    /// directory entry), sorted (test-support for R4c obligation retention).
    #[cfg(test)]
    pub(crate) fn pending_entry_fences(&self) -> Vec<PathBuf> {
        self.dirty_entry_fences.borrow().iter().cloned().collect()
    }
}

impl NodeStore for DiskStore {
    type Error = StoreError;

    fn get(&self, hash: &Hash) -> Result<Option<Arc<Node>>, Self::Error> {
        self.read_node(hash)
    }

    fn put(&mut self, node: &Node) -> Result<Hash, Self::Error> {
        self.write_node(node)
    }

    fn sync_dirty_dirs(&self) -> Result<(), Self::Error> {
        self.sync_dirty_directories()
    }
}

impl DeleteNode for DiskStore {
    type Error = StoreError;

    fn delete(&self, hash: &Hash) -> Result<(), Self::Error> {
        self.delete_node(hash)
    }
}

#[cfg(test)]
mod dirsync_tests {
    use super::{DiskStore, NodeStore};
    use crate::tree::{LeafNode, Node, NodeError};

    fn leaf(key: &[u8], value: &[u8]) -> Result<Node, NodeError> {
        LeafNode::new(vec![(key.to_vec(), value.to_vec())]).map(Node::Leaf)
    }

    /// A freshly persisted node records its DISTINCT parent subdirectory as
    /// dirty, deduped: many nodes landing in the same prefix dir record that dir
    /// exactly once, and nodes in different prefixes are all tracked.
    #[test]
    fn put_records_distinct_parent_dirs_deduped() -> Result<(), Box<dyn std::error::Error>> {
        let temp = tempfile::tempdir()?;
        let mut store = DiskStore::with_cache_capacity(temp.path(), 64)?;

        // Drive many distinct content hashes; each lands in the subdir named by
        // its 2-hex-char prefix. We assert the dirty set is the EXACT distinct
        // set of parent dirs of the files actually written.
        let mut expected: std::collections::BTreeSet<std::path::PathBuf> =
            std::collections::BTreeSet::new();
        for index in 0..64u32 {
            let node = leaf(format!("k{index}").as_bytes(), b"v")?;
            let hash = store.put(&node)?;
            let path = store.node_path(&hash);
            let parent = path
                .parent()
                .ok_or("node path must have a parent")?
                .to_path_buf();
            expected.insert(parent);
        }

        let pending = store.pending_dirty_dirs();
        // Deduped: no parent dir appears twice.
        let unique: std::collections::BTreeSet<_> = pending.iter().cloned().collect();
        assert_eq!(pending.len(), unique.len(), "dirty dirs must be deduped");
        // Exactly the distinct set of parent dirs of every written node.
        assert_eq!(unique, expected);
        Ok(())
    }

    /// The barrier fsyncs and CLEARS the dirty set: after `sync_dirty_dirs` the
    /// pending set is empty (so the next commit starts a fresh batch), and a
    /// re-put of an already-stored node records NOTHING (its directory entry is
    /// already durable — `write_compressed_node` returns `None`).
    #[test]
    fn barrier_clears_dirty_set_and_duplicate_put_records_nothing()
    -> Result<(), Box<dyn std::error::Error>> {
        let temp = tempfile::tempdir()?;
        let mut store = DiskStore::with_cache_capacity(temp.path(), 64)?;

        let node = leaf(b"only", b"value")?;
        store.put(&node)?;
        assert_eq!(
            store.pending_dirty_dirs().len(),
            1,
            "one fresh node, one dir"
        );

        // The barrier fsyncs the dir entry and clears the batch.
        store.sync_dirty_dirs()?;
        assert!(
            store.pending_dirty_dirs().is_empty(),
            "barrier must clear the dirty set"
        );

        // Re-putting an already-stored node persists no new file, so it records
        // no dirty dir (its directory entry is already durable).
        store.put(&node)?;
        assert!(
            store.pending_dirty_dirs().is_empty(),
            "duplicate put must not re-dirty an already-durable dir entry"
        );
        Ok(())
    }

    /// A FAILED barrier keeps the un-fenced dirs pending (adversarial-review
    /// major): a dir leaves the set only after its fsync succeeds, so the
    /// documented "barrier failed, retry the commit" path re-fences the same
    /// dirs instead of silently forgetting them and letting a durable record
    /// name roots whose directory entries were never fenced.
    #[test]
    fn failed_barrier_retains_unfenced_dirs() -> Result<(), Box<dyn std::error::Error>> {
        let temp = tempfile::tempdir()?;
        let mut store = DiskStore::with_cache_capacity(temp.path(), 64)?;

        let node = leaf(b"fence", b"me")?;
        store.put(&node)?;
        let pending = store.pending_dirty_dirs();
        assert_eq!(pending.len(), 1, "one fresh node, one dirty dir");
        let dir = pending
            .first()
            .ok_or("dirty set must hold the node's parent dir")?
            .clone();

        // Remove the recorded dir behind the store's back so its fsync fails.
        std::fs::remove_dir_all(&dir)?;
        assert!(
            store.sync_dirty_dirs().is_err(),
            "fsyncing a removed dir must fail"
        );
        // The failure must NOT have consumed the pending set: a second
        // barrier attempt still fails on the same dir rather than silently
        // succeeding over an emptied set.
        assert_eq!(
            store.pending_dirty_dirs(),
            vec![dir],
            "a failed barrier must retain the un-fenced dir"
        );
        assert!(
            store.sync_dirty_dirs().is_err(),
            "the retry must re-attempt the retained dir, not succeed on empty"
        );
        Ok(())
    }
}