haematite 0.6.2

//! Reopen-after-restart end-to-end (LEDGER A1 stage 4; brief §13's
//! reopen-after-simulated-restart item, §6(c), §9).
//!
//! One flow, cold-restart in the middle: a named branch is created over a
//! real `DiskStore`, advanced through volatile and durable commits, then
//! every in-memory structure is dropped. Recovery reopens the directories
//! with a FRESH registry, `open_branch`es the name, and must find the
//! recorded anchor and the last DURABLE head (volatile advances are volatile,
//! full stop — §4). The reopened state then survives a prune with zero
//! startup re-registration ceremony (§6(c)), and a three-way merge still
//! finds its ancestor — the recorded fork anchor — proving the anchor pin
//! (`protected_roots` = anchors ∪ heads) is doing the §2.3 job it exists for.

use std::fmt;

use super::commit::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use super::conflict::ConflictPolicy;
use super::handle::{BranchError, DEFAULT_SHARD_ID};
use super::lifecycle::{create_branch, open_branch};
use super::merge::{MergeError, merge_with_report};
use super::prune::{PruneError, prune};
use super::refstore::{BranchRefError, BranchRefStore};
use super::registry::BranchRegistry;
use super::snapshot::{SnapshotError, SnapshotRegistry};
use crate::store::{DiskStore, NodeStore, StoreError};
use crate::tree::{Cursor, Hash, LeafNode, Node, NodeError, TreeError, insert};

/// Aggregate error for this test, per the house Result-returning idiom.
#[derive(Debug)]
enum TestError {
    Commit(BranchCommitError),
    Refs(BranchRefError),
    Prune(PruneError),
    Snapshot(SnapshotError),
    Merge(MergeError),
    Branch(BranchError),
    Store(StoreError),
    Tree(TreeError),
    Node(NodeError),
    Io(std::io::Error),
    Missing(&'static str),
}

impl fmt::Display for TestError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Commit(error) => write!(formatter, "commit error: {error}"),
            Self::Refs(error) => write!(formatter, "ref store error: {error}"),
            Self::Prune(error) => write!(formatter, "prune error: {error}"),
            Self::Snapshot(error) => write!(formatter, "snapshot error: {error}"),
            Self::Merge(error) => write!(formatter, "merge error: {error}"),
            Self::Branch(error) => write!(formatter, "branch error: {error}"),
            Self::Store(error) => write!(formatter, "disk store error: {error}"),
            Self::Tree(error) => write!(formatter, "tree error: {error}"),
            Self::Node(error) => write!(formatter, "node error: {error}"),
            Self::Io(error) => write!(formatter, "I/O error: {error}"),
            Self::Missing(what) => write!(formatter, "missing: {what}"),
        }
    }
}

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

impl From<BranchCommitError> for TestError {
    fn from(error: BranchCommitError) -> Self {
        Self::Commit(error)
    }
}

impl From<BranchRefError> for TestError {
    fn from(error: BranchRefError) -> Self {
        Self::Refs(error)
    }
}

impl From<PruneError> for TestError {
    fn from(error: PruneError) -> Self {
        Self::Prune(error)
    }
}

impl From<SnapshotError> for TestError {
    fn from(error: SnapshotError) -> Self {
        Self::Snapshot(error)
    }
}

impl From<MergeError> for TestError {
    fn from(error: MergeError) -> Self {
        Self::Merge(error)
    }
}

impl From<BranchError> for TestError {
    fn from(error: BranchError) -> Self {
        Self::Branch(error)
    }
}

impl From<StoreError> for TestError {
    fn from(error: StoreError) -> Self {
        Self::Store(error)
    }
}

impl From<TreeError> for TestError {
    fn from(error: TreeError) -> Self {
        Self::Tree(error)
    }
}

impl From<NodeError> for TestError {
    fn from(error: NodeError) -> Self {
        Self::Node(error)
    }
}

impl From<std::io::Error> for TestError {
    fn from(error: std::io::Error) -> Self {
        Self::Io(error)
    }
}

fn read_at(store: &DiskStore, root: Hash, key: &[u8]) -> Result<Option<Vec<u8>>, TestError> {
    Cursor::new(store, root).get(key).map_err(TestError::from)
}

/// Session 1: build the base trees, create the named branch, advance it
/// through a volatile checkpoint and two durable commits, then "crash" —
/// every in-memory structure (store, ref store, registry, handle) drops when
/// this returns. Yields `(anchor, junk_root, durable_head)`.
fn session_before_the_crash(
    store_dir: &std::path::Path,
    refs_dir: &std::path::Path,
) -> Result<(Hash, Hash, Hash), TestError> {
    let mut store = DiskStore::new(store_dir)?;
    let empty = store.put(&Node::Leaf(LeafNode::new(Vec::new())?))?;
    let anchor = insert(&mut store, empty, b"base", b"shared history")?;
    // A doomed sibling root for the post-restart prune: reaches its own
    // junk leaf, protected by nothing durable.
    let junk_root = insert(&mut store, anchor, b"junk", b"reclaim me")?;
    store.sync_dirty_dirs()?;

    let mut refs = BranchRefStore::open(refs_dir)?;
    let registry = BranchRegistry::new();
    let branch = create_branch(
        "sess",
        [(DEFAULT_SHARD_ID, anchor)],
        &mut refs,
        &registry,
        10,
    )?;

    // norn's cadence shape (§16.4 Q1): volatile checkpoint, durable
    // commit, durable commit — recovery must land on the LAST DURABLE.
    branch.put(DEFAULT_SHARD_ID, b"checkpoint", b"intra-cadence")?;
    commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Volatile,
            extra_parents: &[],
            timestamp: 20,
        },
    )?;
    for (round, key) in [b"first".as_slice(), b"second".as_slice()]
        .into_iter()
        .enumerate()
    {
        branch.put(DEFAULT_SHARD_ID, key, b"landed")?;
        let timestamp = 30 + u64::try_from(round).unwrap_or(u64::MAX);
        commit_branch(
            &branch,
            &mut store,
            &registry,
            CommitRequest {
                durability: CommitDurability::Durable { refs: &mut refs },
                extra_parents: &[],
                timestamp,
            },
        )?;
    }
    Ok((anchor, junk_root, branch.current_root()))
}

#[test]
fn reopened_branch_survives_a_prune_and_merges_at_its_recorded_ancestor() -> Result<(), TestError> {
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));

    // ---- Session 1: create, advance, "crash". -----------------------------
    let (anchor, junk_root, durable_head) = session_before_the_crash(&store_dir, &refs_dir)?;

    // ---- Session 2: cold recovery. ----------------------------------------
    let mut store = DiskStore::new(&store_dir)?;
    let refs = BranchRefStore::open(&refs_dir)?;
    let registry = BranchRegistry::new();
    let reopened = open_branch("sess", &refs, &registry)?;

    // Anchors and heads recover exactly; the volatile checkpoint did not
    // become a head, but its content rode into the durable commits' tree.
    assert_eq!(reopened.fork_point(), anchor, "recorded merge ancestor");
    assert_eq!(reopened.current_root(), durable_head, "last durable head");
    let record = refs.get("sess").ok_or(TestError::Missing("sess record"))?;
    assert_eq!(record.seq, 2);
    assert_eq!(
        reopened.get(DEFAULT_SHARD_ID, &store, b"checkpoint")?,
        Some(b"intra-cadence".to_vec())
    );
    assert_eq!(
        reopened.get(DEFAULT_SHARD_ID, &store, b"second")?,
        Some(b"landed".to_vec())
    );

    // ---- Prune with the fresh registry: zero startup ceremony. ------------
    // "junk" shares its `base` history with the anchor; only its own leaf
    // may go. The branch's anchor AND head are protected by the record
    // alone — nothing re-registered them after the restart.
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name_at("junk", junk_root, 40)?;
    let report = prune(&store, &registry, &refs, &mut snapshots, "junk")?;
    assert!(
        report.node_count > 0,
        "the junk root itself must be reclaimed"
    );
    assert_eq!(
        read_at(&store, anchor, b"base")?,
        Some(b"shared history".to_vec())
    );
    assert_eq!(
        read_at(&store, durable_head, b"second")?,
        Some(b"landed".to_vec())
    );

    // ---- Three-way merge still finds its ancestor. -------------------------
    // The "trunk" advanced independently from the same anchor; merging the
    // reopened branch into it uses the RECORDED fork anchor as the ancestor,
    // whose nodes the prune was required to spare.
    let trunk = insert(&mut store, anchor, b"trunk", b"parallel work")?;
    let report = merge_with_report(
        &mut store,
        trunk,
        reopened.current_root(),
        reopened.fork_point(),
        &ConflictPolicy::Lww,
    )?;
    assert!(
        report.conflicts.is_empty(),
        "disjoint edits must not conflict"
    );
    for (key, value) in [
        (b"base".as_slice(), b"shared history".as_slice()),
        (b"trunk".as_slice(), b"parallel work".as_slice()),
        (b"checkpoint".as_slice(), b"intra-cadence".as_slice()),
        (b"first".as_slice(), b"landed".as_slice()),
        (b"second".as_slice(), b"landed".as_slice()),
    ] {
        assert_eq!(
            read_at(&store, report.merged_root, key)?,
            Some(value.to_vec()),
            "merged tree must carry both lines' keys"
        );
    }
    Ok(())
}