haematite 0.7.0

//! Shared support for the §13 crash-window battery (LEDGER A1 stage 4):
//! the aggregate test error, the failpoint store wrapper that simulates a
//! kill -9 at a chosen §5 step, and the disk-backed tree helpers. Split from
//! `crash_window_tests.rs` / `crash_recovery_tests.rs` so both stay under the
//! module's 500-line cap; compiled only for test builds (see `mod.rs`).

use std::cell::Cell;
use std::fmt;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use super::commit::{BranchCommitError, CommitDurability, CommitRequest};
use super::handle::BranchError;
use super::refstore::{BranchRefError, BranchRefStore};
use crate::store::{DiskStore, MemoryStore, NodeStore, StoreError};
use crate::tree::{Cursor, Hash, LeafNode, Node, NodeError, TreeError, TreePolicy, insert};

/// Aggregate error for the crash battery, per the house Result-returning
/// test idiom.
#[derive(Debug)]
pub(super) enum TestError {
    Commit(BranchCommitError),
    Refs(BranchRefError),
    Store(StoreError),
    Tree(TreeError),
    Node(NodeError),
    Branch(BranchError),
    Io(std::io::Error),
    Put(String),
    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::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::Branch(error) => write!(formatter, "branch error: {error}"),
            Self::Io(error) => write!(formatter, "I/O error: {error}"),
            Self::Put(reason) => write!(formatter, "store put error: {reason}"),
            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<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<BranchError> for TestError {
    fn from(error: BranchError) -> Self {
        Self::Branch(error)
    }
}

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

/// The failure a [`FailpointStore`] injects at its configured kill point.
#[derive(Debug)]
pub(super) enum FailpointError {
    /// The simulated kill -9: from here the test abandons all in-memory
    /// state instead of retrying.
    Injected,
    /// A real error from the wrapped disk store.
    Store(StoreError),
}

impl fmt::Display for FailpointError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Injected => write!(formatter, "injected crash point"),
            Self::Store(error) => write!(formatter, "disk store error: {error}"),
        }
    }
}

impl std::error::Error for FailpointError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Store(error) => Some(error),
            Self::Injected => None,
        }
    }
}

/// A real [`DiskStore`] with a configurable kill point: fail the N+1th put
/// (mid-§5-step-2) or fail the barrier (§5 step 3). Everything up to the
/// kill point hits the real disk, so recovery sees exactly the bytes a
/// killed process would have left.
#[derive(Debug)]
pub(super) struct FailpointStore {
    pub(super) inner: DiskStore,
    /// `Some(n)`: allow `n` more puts, then inject. `None`: puts unlimited.
    pub(super) puts_allowed: Cell<Option<usize>>,
    pub(super) fail_barrier: bool,
}

impl NodeStore for FailpointStore {
    type Error = FailpointError;

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

    fn put(&mut self, node: &Node) -> Result<Hash, Self::Error> {
        if let Some(remaining) = self.puts_allowed.get() {
            if remaining == 0 {
                return Err(FailpointError::Injected);
            }
            self.puts_allowed.set(Some(remaining - 1));
        }
        self.inner.put(node).map_err(FailpointError::Store)
    }

    fn sync_dirty_dirs(&self) -> Result<(), Self::Error> {
        if self.fail_barrier {
            return Err(FailpointError::Injected);
        }
        self.inner.sync_dirty_dirs().map_err(FailpointError::Store)
    }
}

/// Builds a tree over any store whose mutate errors surface as [`TreeError`].
pub(super) fn build_tree<S: NodeStore + ?Sized>(
    store: &mut S,
    entries: &[(&[u8], &[u8])],
) -> Result<Hash, TestError> {
    let leaf = LeafNode::new(Vec::new())?;
    let mut root = store
        .put(&Node::Leaf(leaf))
        .map_err(|error| TestError::Put(error.to_string()))?;
    for (key, value) in entries {
        root = insert(store, root, key, value, TreePolicy::V1_DEFAULT)?;
    }
    Ok(root)
}

/// The head root the killed commit WOULD have produced, re-derived in an
/// independent in-memory store: history independence makes the hash a pure
/// function of the final key→value set, so this names the on-disk orphan
/// without touching the recovered store's state.
pub(super) fn expected_head(
    base: &[(&[u8], &[u8])],
    extra: &[(&[u8], &[u8])],
) -> Result<Hash, TestError> {
    let mut store = MemoryStore::new();
    let mut root = build_tree(&mut store, base)?;
    for (key, value) in extra {
        root = insert(&mut store, root, key, value, TreePolicy::V1_DEFAULT)?;
    }
    Ok(root)
}

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

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

/// The one `.ref` file in a ref directory (these tests hold one branch each).
pub(super) fn sole_ref_file(refs_dir: &Path) -> Result<PathBuf, TestError> {
    for entry in fs::read_dir(refs_dir)? {
        let path = entry?.path();
        if path.extension().and_then(|extension| extension.to_str()) == Some("ref") {
            return Ok(path);
        }
    }
    Err(TestError::Missing("a .ref file"))
}

pub(super) const BASE: &[(&[u8], &[u8])] = &[(b"base", b"tree")];