haematite 0.6.2

//! Typed-precondition and failure-disposition tests for `commit_branch`
//! (§5 step 2's "nothing observable changes on error", §16.2's CAS refusals).
//! Split from `commit_tests.rs` to keep both under the 500-line cap.

use std::fmt;
use std::sync::Arc;

use super::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use crate::branch::fork::{fork, fork_registered};
use crate::branch::handle::DEFAULT_SHARD_ID;
use crate::branch::lifecycle::{create_branch, open_branch, remove_branch};
use crate::branch::refstore::{BranchRefError, BranchRefRecord, BranchRefStore};
use crate::branch::registry::BranchRegistry;
use crate::store::{MemoryStore, NodeStore};
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 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 unregistered_handle_is_refused() {
    let mut store = MemoryStore::new();
    let registry = BranchRegistry::new();
    let branch = fork(hash(1));

    let result = commit_branch(&branch, &mut store, &registry, volatile());
    assert!(matches!(result, Err(BranchCommitError::Unregistered)));
}

#[test]
fn durable_commit_on_anonymous_handle_is_refused() -> TestResult {
    let mut store = MemoryStore::new();
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = fork_registered(hash(1), &registry);

    let result = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20));
    assert!(matches!(result, Err(BranchCommitError::UnnamedBranch)));
    Ok(())
}

#[test]
fn volatile_extra_parents_are_refused() {
    let mut store = MemoryStore::new();
    let registry = BranchRegistry::new();
    let branch = fork_registered(hash(1), &registry);
    let converge = [hash(2)];

    let result = commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Volatile,
            extra_parents: &converge,
            timestamp: 5,
        },
    );
    assert!(matches!(
        result,
        Err(BranchCommitError::VolatileExtraParents)
    ));
}

#[test]
fn stale_seq_between_two_handles_is_typed() -> 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 first = create_branch(
        "shared",
        [(DEFAULT_SHARD_ID, root)],
        &mut refs,
        &registry,
        10,
    )?;
    let second = open_branch("shared", &refs, &registry)?;

    first.put(DEFAULT_SHARD_ID, b"key", b"first")?;
    commit_branch(&first, &mut store, &registry, durable(&mut refs, 20))?;

    second.put(DEFAULT_SHARD_ID, b"key", b"second")?;
    let result = commit_branch(&second, &mut store, &registry, durable(&mut refs, 30));
    assert!(matches!(
        result,
        Err(BranchCommitError::Ref(BranchRefError::StaleSeq {
            expected: 0,
            found: 1,
            ..
        }))
    ));
    // Refused, not half-applied: the stale handle's buffer and root survive.
    assert!(second.shard_is_dirty(DEFAULT_SHARD_ID)?);
    assert_eq!(second.current_root(), root);
    Ok(())
}

#[test]
fn generation_mismatch_after_recreate_is_typed() -> TestResult {
    // §16.2's ABA: a stale handle from a removed generation of the same name
    // would pass a seq-only CAS (both generations start at 0); the creation
    // identity refuses it.
    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 stale = create_branch(
        "lease",
        [(DEFAULT_SHARD_ID, root)],
        &mut refs,
        &registry,
        100,
    )?;
    remove_branch("lease", &mut refs)?;
    let _fresh = create_branch(
        "lease",
        [(DEFAULT_SHARD_ID, root)],
        &mut refs,
        &registry,
        200,
    )?;

    stale.put(DEFAULT_SHARD_ID, b"key", b"hijack")?;
    let result = commit_branch(&stale, &mut store, &registry, durable(&mut refs, 300));
    assert!(matches!(
        result,
        Err(BranchCommitError::Ref(
            BranchRefError::BranchGenerationMismatch {
                expected_created: 100,
                found_created: 200,
                ..
            }
        ))
    ));
    Ok(())
}

#[test]
fn commit_after_remove_is_branch_removed() -> 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)?;

    branch.put(DEFAULT_SHARD_ID, b"key", b"orphan")?;
    let result = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20));
    assert!(matches!(
        result,
        Err(BranchCommitError::Ref(BranchRefError::BranchRemoved(name))) if name == "gone"
    ));
    Ok(())
}

/// Error injected by [`FailingStore`].
#[derive(Debug)]
struct StoreDown;

impl fmt::Display for StoreDown {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(formatter, "store down")
    }
}

impl std::error::Error for StoreDown {}

/// A store wrapper that fails on demand, for §5's failure-disposition tests.
#[derive(Debug)]
struct FailingStore {
    inner: MemoryStore,
    fail_puts: bool,
    fail_barrier: bool,
}

impl NodeStore for FailingStore {
    type Error = StoreDown;

    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> {
        if self.fail_puts {
            Err(StoreDown)
        } else {
            Ok(MemoryStore::put(&mut self.inner, node))
        }
    }

    fn sync_dirty_dirs(&self) -> Result<(), Self::Error> {
        if self.fail_barrier {
            Err(StoreDown)
        } else {
            Ok(())
        }
    }
}

#[test]
fn failed_materialisation_changes_nothing_and_is_retryable() -> TestResult {
    let mut inner = MemoryStore::new();
    let root = build_tree(&mut inner, &[(b"base", b"tree")])?;
    let mut store = FailingStore {
        inner,
        fail_puts: true,
        fail_barrier: false,
    };
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch(
        "flaky",
        [(DEFAULT_SHARD_ID, root)],
        &mut refs,
        &registry,
        10,
    )?;
    let live_before = registry.live_roots();

    branch.put(DEFAULT_SHARD_ID, b"key", b"value")?;
    let result = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20));
    assert!(matches!(result, Err(BranchCommitError::Tree(_))));

    // Nothing observable changed: buffer intact, root unmoved, record at
    // seq 0, registry pins exactly as before.
    assert!(branch.shard_is_dirty(DEFAULT_SHARD_ID)?);
    assert_eq!(branch.current_root(), root);
    assert_eq!(get_record(&refs, "flaky")?.seq, 0);
    assert_eq!(registry.live_roots(), live_before);

    // Retryable: the same commit succeeds once the store recovers.
    store.fail_puts = false;
    let outcome = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 30))?;
    assert_eq!(outcome.seq, Some(1));
    assert!(!branch.shard_is_dirty(DEFAULT_SHARD_ID)?);
    Ok(())
}

#[test]
fn failed_barrier_changes_nothing_and_is_retryable() -> TestResult {
    // The barrier fails AFTER materialisation succeeded: the freshly written
    // nodes are orphans (leak-safe), and every observable — buffer, root,
    // record, pins — is exactly as before the call.
    let mut inner = MemoryStore::new();
    let root = build_tree(&mut inner, &[(b"base", b"tree")])?;
    let mut store = FailingStore {
        inner,
        fail_puts: false,
        fail_barrier: true,
    };
    let registry = BranchRegistry::new();
    let dir = tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch(
        "fenced",
        [(DEFAULT_SHARD_ID, root)],
        &mut refs,
        &registry,
        10,
    )?;
    let live_before = registry.live_roots();

    branch.put(DEFAULT_SHARD_ID, b"key", b"value")?;
    let result = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20));
    assert!(matches!(result, Err(BranchCommitError::Barrier(_))));

    assert!(branch.shard_is_dirty(DEFAULT_SHARD_ID)?);
    assert_eq!(branch.current_root(), root);
    assert_eq!(get_record(&refs, "fenced")?.seq, 0);
    assert_eq!(registry.live_roots(), live_before);

    store.fail_barrier = false;
    let outcome = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 30))?;
    assert_eq!(outcome.seq, Some(1));
    Ok(())
}