haematite 0.6.1

//! §14.3(b): the barrier-counting store decorator (LEDGER A1 stage 4).
//!
//! Proves the §5 step-3/step-4 ordering from the store's point of view:
//! `sync_dirty_dirs` runs EXACTLY ONCE per durable commit, STRICTLY BEFORE
//! the ref-record install. The decorator does not merely count: at every
//! barrier it reopens the branch's ref directory cold and records the ON-DISK
//! sequence it finds. A commit that installed the record first would be seen
//! holding the NEW seq at barrier time; a commit that fenced twice grows the
//! observation log by two. Volatile commits and durable no-ops must not fence
//! at all (§5 step 3 "skipped for Volatile"; the §14.2 no-op returns before
//! step 3).

use std::cell::RefCell;
use std::convert::Infallible;
use std::path::PathBuf;
use std::sync::Arc;

use super::commit::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use super::handle::DEFAULT_SHARD_ID;
use super::lifecycle::create_branch;
use super::refstore::{BranchRefError, BranchRefStore};
use super::registry::BranchRegistry;
use crate::store::{MemoryStore, NodeStore};
use crate::tree::{Hash, LeafNode, Node, TreeError, insert};

type TestResult = Result<(), BranchCommitError>;

const BRANCH: &str = "spied";

fn build_tree(store: &mut MemoryStore, entries: &[(&[u8], &[u8])]) -> Result<Hash, TreeError> {
    let leaf = LeafNode::new(Vec::new())?;
    let mut root = store.put(&Node::Leaf(leaf));
    for (key, value) in entries {
        root = insert(store, root, key, value)?;
    }
    Ok(root)
}

fn tempdir() -> Result<tempfile::TempDir, BranchCommitError> {
    tempfile::tempdir().map_err(|error| BranchCommitError::Ref(BranchRefError::Io(error)))
}

/// A [`NodeStore`] decorator whose barrier spies on the durable world: each
/// `sync_dirty_dirs` call appends the branch's CURRENT on-disk record seq
/// (read via a cold `BranchRefStore::open` over the ref directory, so no
/// in-memory state can flatter the answer). `None` = no readable record.
#[derive(Debug)]
struct BarrierSpyStore {
    inner: MemoryStore,
    refs_dir: PathBuf,
    barrier_seqs: RefCell<Vec<Option<u64>>>,
}

impl BarrierSpyStore {
    fn new(inner: MemoryStore, refs_dir: PathBuf) -> Self {
        Self {
            inner,
            refs_dir,
            barrier_seqs: RefCell::new(Vec::new()),
        }
    }

    fn barrier_seqs(&self) -> Vec<Option<u64>> {
        self.barrier_seqs.borrow().clone()
    }
}

impl NodeStore for BarrierSpyStore {
    type Error = Infallible;

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

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

    fn sync_dirty_dirs(&self) -> Result<(), Self::Error> {
        let on_disk_seq = BranchRefStore::open(&self.refs_dir)
            .ok()
            .and_then(|refs| refs.get(BRANCH).map(|record| record.seq));
        self.barrier_seqs.borrow_mut().push(on_disk_seq);
        Ok(())
    }
}

#[test]
fn barrier_runs_exactly_once_strictly_before_the_install_per_durable_commit() -> TestResult {
    let mut base = MemoryStore::new();
    let root = build_tree(&mut base, &[(b"base", b"tree")])?;
    let dir = tempdir()?;
    let mut store = BarrierSpyStore::new(base, dir.path().to_path_buf());
    let registry = BranchRegistry::new();
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch(BRANCH, [(DEFAULT_SHARD_ID, root)], &mut refs, &registry, 10)?;

    for (round, expected_pre_install_seq) in (0u64..3).enumerate() {
        branch.put(DEFAULT_SHARD_ID, b"key", round.to_le_bytes())?;
        let outcome = commit_branch(
            &branch,
            &mut store,
            &registry,
            CommitRequest {
                durability: CommitDurability::Durable { refs: &mut refs },
                extra_parents: &[],
                timestamp: 20 + expected_pre_install_seq,
            },
        )?;
        assert_eq!(outcome.seq, Some(expected_pre_install_seq + 1));

        let observed = store.barrier_seqs();
        // Exactly once per durable commit: one new observation per round.
        assert_eq!(
            observed.len(),
            round + 1,
            "round {round}: exactly one barrier per durable commit"
        );
        // Strictly before the install: at barrier time the durable world
        // still held the PREVIOUS record. Install-then-fence would read the
        // freshly bumped seq here instead.
        assert_eq!(
            observed[round],
            Some(expected_pre_install_seq),
            "round {round}: the barrier must observe the pre-install record"
        );
    }
    Ok(())
}

#[test]
fn volatile_commits_and_durable_noops_never_fence() -> TestResult {
    let mut base = MemoryStore::new();
    let root = build_tree(&mut base, &[(b"base", b"tree")])?;
    let dir = tempdir()?;
    let mut store = BarrierSpyStore::new(base, dir.path().to_path_buf());
    let registry = BranchRegistry::new();
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch(BRANCH, [(DEFAULT_SHARD_ID, root)], &mut refs, &registry, 10)?;

    // Volatile commit with real content: no barrier (§5 step 3).
    branch.put(DEFAULT_SHARD_ID, b"key", b"volatile")?;
    commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Volatile,
            extra_parents: &[],
            timestamp: 20,
        },
    )?;
    assert!(
        store.barrier_seqs().is_empty(),
        "a volatile commit must not fence directory entries"
    );

    // Durable no-op (empty buffer, no extra parents): returns before step 3,
    // installs nothing, fences nothing.
    let outcome = commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[],
            timestamp: 30,
        },
    )?;
    assert!(!outcome.advanced);
    assert!(
        store.barrier_seqs().is_empty(),
        "a durable no-op must not fence directory entries"
    );
    Ok(())
}

#[test]
fn record_only_commit_still_fences_before_its_install() -> TestResult {
    // §14.2's record-only commit installs a record whose heads are unchanged
    // but which a restart will trust; the barrier before it also fences any
    // directory entries left pending by earlier volatile commits (§4), so it
    // must run — once, and before the install — exactly like a content
    // commit.
    let mut base = MemoryStore::new();
    let root = build_tree(&mut base, &[(b"base", b"tree")])?;
    let dir = tempdir()?;
    let mut store = BarrierSpyStore::new(base, dir.path().to_path_buf());
    let registry = BranchRegistry::new();
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch(BRANCH, [(DEFAULT_SHARD_ID, root)], &mut refs, &registry, 10)?;

    let converge = Hash::from_bytes([9; 32]);
    let outcome = commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: std::slice::from_ref(&converge),
            timestamp: 20,
        },
    )?;
    assert!(outcome.advanced);
    assert_eq!(outcome.seq, Some(1));
    assert_eq!(
        store.barrier_seqs(),
        vec![Some(0)],
        "one barrier, observing the pre-install record"
    );
    Ok(())
}