haematite 0.6.1

//! Happy-path unit tests for the branch commit path. Split out of `commit.rs`
//! via `#[path]` so that file stays within the branch module's 500-line cap;
//! typed-refusal and failure-disposition coverage lives in
//! `commit_error_tests.rs`.

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

type TestResult = Result<(), BranchCommitError>;

fn hash(byte: u8) -> Hash {
    Hash::from_bytes([byte; 32])
}

fn empty_root(store: &mut MemoryStore) -> Result<Hash, TreeError> {
    let leaf = LeafNode::new(Vec::new())?;
    Ok(store.put(&Node::Leaf(leaf)))
}

fn build_tree(store: &mut MemoryStore, entries: &[(&[u8], &[u8])]) -> Result<Hash, TreeError> {
    let mut root = empty_root(store)?;
    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)))
}

fn get_record<'refs>(
    refs: &'refs BranchRefStore,
    name: &str,
) -> Result<&'refs BranchRefRecord, BranchCommitError> {
    refs.get(name)
        .ok_or_else(|| BranchCommitError::Ref(BranchRefError::BranchRemoved(name.to_owned())))
}

fn volatile() -> CommitRequest<'static> {
    CommitRequest {
        durability: CommitDurability::Volatile,
        extra_parents: &[],
        timestamp: 1,
    }
}

fn durable(refs: &mut BranchRefStore, timestamp: u64) -> CommitRequest<'_> {
    CommitRequest {
        durability: CommitDurability::Durable { refs },
        extra_parents: &[],
        timestamp,
    }
}

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

    branch.put(DEFAULT_SHARD_ID, b"key", b"volatile")?;
    let outcome = commit_branch(&branch, &mut store, &registry, volatile())?;

    assert!(outcome.advanced);
    assert_eq!(outcome.seq, None);
    let new_root = branch.current_root();
    assert_ne!(new_root, root);
    assert_eq!(outcome.heads, vec![(DEFAULT_SHARD_ID, new_root)]);
    // The buffer drained into the tree: the value now reads through the root.
    assert!(!branch.shard_is_dirty(DEFAULT_SHARD_ID)?);
    assert_eq!(
        Cursor::new(&store, new_root).get(b"key")?,
        Some(b"volatile".to_vec())
    );
    // In memory only: the durable record still holds the anchor at seq 0.
    let record = get_record(&refs, "wip")?;
    assert_eq!(record.seq, 0);
    assert_eq!(record.shards[0].head, root);
    // Pins advanced under one lock: new head live, anchor-role pin intact.
    let live = registry.live_roots();
    assert!(live.contains(&new_root));
    assert!(live.contains(&root));
    Ok(())
}

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

    branch.put(DEFAULT_SHARD_ID, b"key", b"durable")?;
    let outcome = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20))?;

    assert!(outcome.advanced);
    assert_eq!(outcome.seq, Some(1));
    let head = branch.current_root();
    let record = get_record(&refs, "sess")?;
    assert_eq!(record.seq, 1);
    assert_eq!(record.timestamp, 20);
    assert_eq!(record.shards[0].fork_anchor, root);
    assert_eq!(record.shards[0].head, head);
    assert_eq!(record.parents, vec![root]);
    // Restart-visible: a fresh open of the directory sees the same record.
    let reopened = BranchRefStore::open(dir.path())?;
    assert_eq!(reopened.get("sess"), Some(record));

    // The binding seq was bumped in step 6: the next durable commit passes
    // the CAS and chains parents[0] to this record's head.
    branch.put(DEFAULT_SHARD_ID, b"key2", b"again")?;
    let second = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 30))?;
    assert_eq!(second.seq, Some(2));
    assert_eq!(get_record(&refs, "sess")?.parents, vec![head]);
    Ok(())
}

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

    let quiet = commit_branch(&branch, &mut store, &registry, volatile())?;
    assert!(!quiet.advanced);
    assert_eq!(quiet.seq, None);
    assert_eq!(quiet.heads, vec![(DEFAULT_SHARD_ID, root)]);

    let still = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20))?;
    assert!(!still.advanced);
    assert_eq!(still.seq, Some(0));
    // No record installed: seq and timestamp are untouched.
    let record = get_record(&refs, "idle")?;
    assert_eq!(record.seq, 0);
    assert_eq!(record.timestamp, 10);
    Ok(())
}

#[test]
fn record_only_commit_self_parents_and_advances_seq() -> TestResult {
    // §14.2 + §16.3: empty buffers + converge links is NOT a no-op — the
    // record advances with unchanged heads and an explicit self-parent.
    let mut store = MemoryStore::new();
    let root = build_tree(&mut store, &[(b"base", b"tree")])?;
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch("dag", [(DEFAULT_SHARD_ID, root)], &mut refs, &registry, 10)?;

    let converge = hash(9);
    let outcome = commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[converge],
            timestamp: 20,
        },
    )?;

    assert!(outcome.advanced);
    assert_eq!(outcome.seq, Some(1));
    assert_eq!(outcome.heads, vec![(DEFAULT_SHARD_ID, root)]);
    let record = get_record(&refs, "dag")?;
    assert_eq!(record.seq, 1);
    assert_eq!(
        record.shards[0].head, root,
        "record-only heads are unchanged"
    );
    assert_eq!(record.parents, vec![root, converge]);
    assert_eq!(branch.current_root(), root);
    Ok(())
}

#[test]
fn durable_after_volatile_parents_are_previous_record_heads() -> TestResult {
    // §16.3 uniform parents rule: lineage is durable-to-durable, skipping
    // volatile intermediates — never the in-memory current_root.
    let mut store = MemoryStore::new();
    let root = build_tree(&mut store, &[(b"base", b"tree")])?;
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch(
        "cadence",
        [(DEFAULT_SHARD_ID, root)],
        &mut refs,
        &registry,
        10,
    )?;

    branch.put(DEFAULT_SHARD_ID, b"first", b"checkpoint")?;
    commit_branch(&branch, &mut store, &registry, volatile())?;
    let volatile_root = branch.current_root();
    assert_ne!(volatile_root, root);

    branch.put(DEFAULT_SHARD_ID, b"second", b"durable")?;
    let outcome = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 30))?;

    assert_eq!(outcome.seq, Some(1));
    let record = get_record(&refs, "cadence")?;
    assert_eq!(
        record.parents,
        vec![root],
        "parents[0] must be the previous RECORD head, not the volatile root"
    );
    assert_eq!(record.shards[0].head, branch.current_root());
    assert_ne!(record.shards[0].head, volatile_root);
    // The durable head still reaches the volatile-committed content.
    assert_eq!(
        Cursor::new(&store, record.shards[0].head).get(b"first")?,
        Some(b"checkpoint".to_vec())
    );
    Ok(())
}

#[test]
fn multi_shard_commit_swaps_all_heads_in_one_record() -> TestResult {
    let mut store = MemoryStore::new();
    let root_zero = build_tree(&mut store, &[(b"zero", b"a")])?;
    let root_three = build_tree(&mut store, &[(b"three", b"b")])?;
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch(
        "multi",
        [(0, root_zero), (3, root_three)],
        &mut refs,
        &registry,
        10,
    )?;

    branch.put(0, b"zero-key", b"zero-value")?;
    branch.put(3, b"three-key", b"three-value")?;
    let outcome = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20))?;

    assert_eq!(outcome.seq, Some(1));
    let record = get_record(&refs, "multi")?;
    assert_eq!(record.seq, 1);
    assert_eq!(record.shards.len(), 2);
    // One record covers every shard: both heads advanced past their anchors.
    assert_eq!(record.shards[0].shard_id, 0);
    assert_ne!(record.shards[0].head, root_zero);
    assert_eq!(record.shards[0].fork_anchor, root_zero);
    assert_eq!(record.shards[1].shard_id, 3);
    assert_ne!(record.shards[1].head, root_three);
    assert_eq!(record.shards[1].fork_anchor, root_three);
    // Both durable parents: the previous composite heads, ascending shard id.
    assert_eq!(record.parents, vec![root_zero, root_three]);
    // In-memory heads match the record, ascending shard id.
    assert_eq!(
        outcome.heads,
        vec![(0, record.shards[0].head), (3, record.shards[1].head)]
    );
    assert_eq!(branch.shard_current_root(0), Some(record.shards[0].head));
    assert_eq!(branch.shard_current_root(3), Some(record.shards[1].head));
    Ok(())
}

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

    branch.put(DEFAULT_SHARD_ID, b"one", b"1")?;
    commit_branch(&branch, &mut store, &registry, volatile())?;
    let first_head = branch.current_root();
    branch.put(DEFAULT_SHARD_ID, b"two", b"2")?;
    commit_branch(&branch, &mut store, &registry, volatile())?;
    let second_head = branch.current_root();

    // The head-role pin advanced twice; the anchor-role pin never moved
    // (§14.1: no first-advance special case, the refcount does the work).
    let live = registry.live_roots();
    assert!(live.contains(&root), "anchor must stay pinned for merge");
    assert!(live.contains(&second_head));
    assert!(
        !live.contains(&first_head),
        "superseded head must be released"
    );

    drop(branch);
    assert!(registry.live_roots().is_empty());
    Ok(())
}