haematite 0.6.2

//! Unit tests for the branch lifecycle (`fork_from` / `fork_at` / create /
//! open / remove). Split out of `lifecycle.rs` via `#[path]` per the module's
//! 500-line cap convention.

use super::{create_branch, fork_at, fork_from, open_branch, remove_branch};
use crate::branch::commit::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use crate::branch::fork::fork_registered;
use crate::branch::handle::DEFAULT_SHARD_ID;
use crate::branch::refstore::{BranchRefError, BranchRefStore};
use crate::branch::registry::BranchRegistry;
use crate::branch::snapshot::{SnapshotError, SnapshotRegistry};
use crate::store::MemoryStore;
use crate::tree::{Hash, LeafNode, Node, TreeError, insert};

type TestResult = Result<(), BranchCommitError>;

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

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)))
}

fn snapshot_error(error: &SnapshotError) -> BranchCommitError {
    BranchCommitError::Barrier(error.to_string())
}

#[test]
fn fork_from_anchors_at_committed_heads_buffer_invisible() -> TestResult {
    let mut store = MemoryStore::new();
    let root = build_tree(&mut store, &[(b"key", b"committed")])?;
    let registry = BranchRegistry::new();
    let parent = fork_registered(root, &registry);

    parent.put(DEFAULT_SHARD_ID, b"key", b"uncommitted")?;
    let child = fork_from(&parent, &registry)?;

    // Anchored at the committed head: buffered intent is not a tree and has
    // no hash to anchor to (§9).
    assert_eq!(child.fork_point(), root);
    assert_eq!(child.current_root(), root);
    assert_eq!(
        child.get(DEFAULT_SHARD_ID, &store, b"key")?,
        Some(b"committed".to_vec())
    );

    // The child's own pins keep the anchor live after the parent is gone.
    drop(parent);
    assert!(registry.live_roots().contains(&root));
    drop(child);
    assert!(registry.live_roots().is_empty());
    Ok(())
}

#[test]
fn fork_from_sees_parent_committed_advances() -> 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 parent = create_branch(
        "parent",
        [(DEFAULT_SHARD_ID, root)],
        &mut refs,
        &registry,
        10,
    )?;

    parent.put(DEFAULT_SHARD_ID, b"landed", b"yes")?;
    commit_branch(
        &parent,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Volatile,
            extra_parents: &[],
            timestamp: 20,
        },
    )?;
    let committed = parent.current_root();
    parent.put(DEFAULT_SHARD_ID, b"pending", b"no")?;

    let child = fork_from(&parent, &registry)?;
    assert_eq!(child.fork_point(), committed);
    assert_eq!(
        child.get(DEFAULT_SHARD_ID, &store, b"landed")?,
        Some(b"yes".to_vec())
    );
    assert_eq!(child.get(DEFAULT_SHARD_ID, &store, b"pending")?, None);
    Ok(())
}

#[test]
fn create_branch_pins_and_installs_seq_zero_record() -> TestResult {
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;

    let branch = create_branch(
        "fresh",
        [(0, hash(1)), (3, hash(2))],
        &mut refs,
        &registry,
        42,
    )?;

    assert!(branch.is_named());
    let record = refs
        .get("fresh")
        .ok_or_else(|| BranchCommitError::Ref(BranchRefError::BranchRemoved("fresh".into())))?;
    assert_eq!(record.created, 42);
    assert_eq!(record.seq, 0);
    assert_eq!(record.shards[0].fork_anchor, hash(1));
    assert_eq!(record.shards[0].head, hash(1));
    assert_eq!(record.shards[1].fork_anchor, hash(2));
    assert_eq!(record.shards[1].head, hash(2));
    assert!(record.parents.is_empty());
    let live = registry.live_roots();
    assert!(live.contains(&hash(1)));
    assert!(live.contains(&hash(2)));

    drop(branch);
    // In-memory pins release with the guard; the record still protects the
    // roots durably.
    assert!(registry.live_roots().is_empty());
    assert!(refs.protected_roots().contains(&hash(1)));
    Ok(())
}

#[test]
fn duplicate_create_is_typed_and_releases_its_pins() -> TestResult {
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let _first = create_branch("dup", [(0, hash(1))], &mut refs, &registry, 10)?;
    let live_before = registry.live_roots();

    let result = create_branch("dup", [(0, hash(2))], &mut refs, &registry, 20);
    assert!(matches!(
        result,
        Err(BranchCommitError::Ref(BranchRefError::DuplicateBranch(name))) if name == "dup"
    ));
    // The refused create's speculative pins were released with its guard.
    assert_eq!(registry.live_roots(), live_before);
    Ok(())
}

#[test]
fn open_branch_restores_anchor_as_fork_point() -> 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 head = {
        let branch = create_branch("sess", [(DEFAULT_SHARD_ID, root)], &mut refs, &registry, 10)?;
        branch.put(DEFAULT_SHARD_ID, b"key", b"advanced")?;
        commit_branch(
            &branch,
            &mut store,
            &registry,
            CommitRequest {
                durability: CommitDurability::Durable { refs: &mut refs },
                extra_parents: &[],
                timestamp: 20,
            },
        )?;
        branch.current_root()
    };
    // Simulated restart: the in-memory registry starts empty, the ref store
    // reopens from disk.
    assert!(registry.live_roots().is_empty());
    let refs = BranchRefStore::open(dir.path())?;

    let reopened = open_branch("sess", &refs, &registry)?;
    assert!(reopened.is_named());
    assert_eq!(
        reopened.fork_point(),
        root,
        "the recorded anchor — the merge ancestor — must survive reopen"
    );
    assert_eq!(reopened.current_root(), head);
    let live = registry.live_roots();
    assert!(live.contains(&root));
    assert!(live.contains(&head));

    // The binding carries the recorded seq: the next durable commit passes
    // the CAS (a fresh mutable ref store over the same directory).
    let mut refs = BranchRefStore::open(dir.path())?;
    reopened.put(DEFAULT_SHARD_ID, b"key2", b"next")?;
    let outcome = commit_branch(
        &reopened,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[],
            timestamp: 30,
        },
    )?;
    assert_eq!(outcome.seq, Some(2));
    Ok(())
}

#[test]
fn open_unknown_branch_is_typed() -> TestResult {
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;

    let result = open_branch("never", &refs, &registry);
    assert!(matches!(
        result,
        Err(BranchCommitError::Ref(BranchRefError::BranchRemoved(name))) if name == "never"
    ));
    Ok(())
}

#[test]
fn remove_branch_returns_record_and_leaves_unknown_names_none() -> TestResult {
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    create_branch("done", [(0, hash(1))], &mut refs, &registry, 10)?;

    let removed = remove_branch("done", &mut refs).map_err(BranchCommitError::Ref)?;
    assert!(matches!(removed, Some(record) if record.name == "done"));
    let again = remove_branch("done", &mut refs).map_err(BranchCommitError::Ref)?;
    assert!(again.is_none());
    Ok(())
}

#[test]
fn fork_at_unprotected_hash_is_hard_refused() -> TestResult {
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let snapshots = SnapshotRegistry::new();
    let rogue = hash(9);

    let result = fork_at(rogue, &registry, &refs, &snapshots);
    let Err(error) = result else {
        return Err(BranchCommitError::UnprotectedAnchor { root: rogue });
    };
    assert!(matches!(
        error,
        BranchCommitError::UnprotectedAnchor { root } if root == rogue
    ));
    // The message names the recipe (§16.4 Q3): snapshot first, then fork.
    assert!(error.to_string().contains("name a snapshot"));
    // Refusal registered nothing.
    assert!(registry.live_roots().is_empty());
    Ok(())
}

#[test]
fn fork_at_accepts_each_protection_source() -> TestResult {
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();

    // (a) Named-snapshot protection.
    snapshots
        .name("pin", hash(1))
        .map_err(|error| snapshot_error(&error))?;
    let from_snapshot = fork_at(hash(1), &registry, &refs, &snapshots)?;
    assert_eq!(from_snapshot.fork_point(), hash(1));

    // (b) Durable ref-record protection (anchor/head of a named branch).
    let named = create_branch("layer", [(0, hash(2))], &mut refs, &registry, 10)?;
    let from_record = fork_at(hash(2), &registry, &refs, &snapshots)?;
    assert_eq!(from_record.fork_point(), hash(2));

    // (c) Live registry protection: pinned only by another handle, and the
    // check+register under one lock means the new fork keeps it live even
    // once that handle drops.
    let holder = fork_registered(hash(3), &registry);
    let from_live = fork_at(hash(3), &registry, &refs, &snapshots)?;
    drop(holder);
    assert!(registry.live_roots().contains(&hash(3)));

    drop((named, from_snapshot, from_record, from_live));
    Ok(())
}

#[test]
fn fork_from_refuses_unregistered_parent() -> TestResult {
    let mut store = MemoryStore::new();
    let root = build_tree(&mut store, &[(b"key", b"v")])?;
    let registry = BranchRegistry::new();

    // BranchHandle::new is unregistered by construction: its heads carry no
    // prune pin, so anchoring a child there is the §16.4 Q3 hazard by
    // another door.
    let parent = crate::branch::handle::BranchHandle::new(root);
    let result = fork_from(&parent, &registry);
    assert!(
        matches!(
            result,
            Err(crate::branch::handle::BranchError::UnregisteredParent)
        ),
        "fork_from must refuse an unregistered parent"
    );
    Ok(())
}

#[test]
fn durable_noop_commit_still_runs_cas() -> 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("gone", [(DEFAULT_SHARD_ID, root)], &mut refs, &registry, 10)?;
    remove_branch("gone", &mut refs).map_err(BranchCommitError::Ref)?;

    // Empty buffer, no extra parents — the documented §14.2 no-op — but the
    // handle is stale on a removed branch: the §16.2 CAS must refuse here
    // too, never return a reassuring Ok carrying a seq the ref store no
    // longer recognises (adversarial-review finding).
    let result = commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[],
            timestamp: 20,
        },
    );
    assert!(
        matches!(
            result,
            Err(BranchCommitError::Ref(BranchRefError::BranchRemoved(ref name))) if name == "gone"
        ),
        "a durable no-op on a removed branch must surface BranchRemoved"
    );
    Ok(())
}