haematite 0.7.0

//! §13 crash-window tests, PRE-rename half (LEDGER A1 stage 4).
//!
//! Each test simulates a kill -9 landing between two numbered §5 steps
//! before the step-4 rename and asserts the §1 crash-window table's
//! disposition after recovery: the OLD record stays valid and everything the
//! dead commit wrote is orphans only. The simulation is the strongest one
//! safe Rust offers: a failpoint store wrapper (see `crash_support.rs`)
//! injects the failure at the chosen step over a REAL `DiskStore`, then the
//! test ABANDONS every in-memory structure (handle, registry, ref store,
//! node store) and recovers cold — reopened directories, a fresh registry,
//! `open_branch` — exactly the restart path. The post-rename windows and the
//! rename-atomicity simulation live in `crash_recovery_tests.rs`.

use std::cell::Cell;

use super::commit::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use super::crash_support::{
    BASE, FailpointStore, TestError, build_tree, durable, expected_head, read_at,
};
use super::lifecycle::{create_branch, open_branch};
use super::refstore::BranchRefStore;
use super::registry::BranchRegistry;
use crate::store::{DiskStore, NodeStore};
use crate::tree::TreePolicy;

#[test]
fn crash_before_commit_discards_buffered_intent() -> Result<(), TestError> {
    // Window before step 2 ever runs: buffered branch writes are volatile by
    // contract (§4) — a crash with a dirty buffer recovers at the anchor,
    // with nothing of the buffered intent anywhere on disk.
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));
    let anchor = {
        let mut store = DiskStore::new(&store_dir)?;
        let anchor = build_tree(&mut store, BASE)?;
        store.sync_dirty_dirs()?;
        let mut refs = BranchRefStore::open(&refs_dir)?;
        let registry = BranchRegistry::new();
        let branch = create_branch("wal", [(0, anchor)], &mut refs, &registry, 10)?;
        branch.put(0, b"pending", b"never committed")?;
        anchor
        // Kill: handle, registry, ref store and node store all drop here.
    };

    let store = DiskStore::new(&store_dir)?;
    let refs = BranchRefStore::open(&refs_dir)?;
    let registry = BranchRegistry::new();
    let reopened = open_branch("wal", &refs, &registry)?;
    assert_eq!(reopened.current_root(), anchor);
    assert_eq!(reopened.fork_point(), anchor);
    assert_eq!(read_at(&store, anchor, b"base")?, Some(b"tree".to_vec()));
    assert_eq!(read_at(&store, anchor, b"pending")?, None);
    Ok(())
}

#[test]
fn crash_mid_materialisation_leaves_old_record_and_orphans_only() -> Result<(), TestError> {
    // Kill inside step 2, on a TWO-shard commit after the first shard's
    // nodes hit the disk: the §1 disposition is old record valid + orphans
    // only, and the record must name BOTH old heads — never a mix (§7).
    const BASE_ZERO: &[(&[u8], &[u8])] = &[(b"zero", b"base")];
    const BASE_THREE: &[(&[u8], &[u8])] = &[(b"three", b"base")];
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));
    let orphan = expected_head(BASE_ZERO, &[(b"zero-new", b"v0")])?;

    let (anchor_zero, anchor_three) = {
        let mut disk = DiskStore::new(&store_dir)?;
        let anchor_zero = build_tree(&mut disk, BASE_ZERO)?;
        let anchor_three = build_tree(&mut disk, BASE_THREE)?;
        disk.sync_dirty_dirs()?;
        let mut refs = BranchRefStore::open(&refs_dir)?;
        let registry = BranchRegistry::new();
        let branch = create_branch(
            "wal",
            [(0, anchor_zero), (3, anchor_three)],
            &mut refs,
            &registry,
            10,
        )?;
        branch.put(0, b"zero-new", b"v0")?;
        branch.put(3, b"three-new", b"v3")?;

        // Shards materialise ascending by id: shard 0 writes its one new
        // leaf, then shard 3's put is the kill point.
        let mut store = FailpointStore {
            inner: disk,
            puts_allowed: Cell::new(Some(1)),
            fail_barrier: false,
        };
        let result = commit_branch(
            &branch,
            &mut store,
            &registry,
            durable(&mut refs, 20),
            TreePolicy::V1_DEFAULT,
        );
        assert!(matches!(result, Err(BranchCommitError::Tree(_))));
        // Shard 0's would-be head is already on disk — the orphan.
        assert!(store.inner.get(&orphan)?.is_some());
        (anchor_zero, anchor_three)
        // Kill: everything drops un-swapped.
    };

    let store = DiskStore::new(&store_dir)?;
    let refs = BranchRefStore::open(&refs_dir)?;
    let registry = BranchRegistry::new();
    let record = refs.get("wal").ok_or(TestError::Missing("wal record"))?;
    // All-or-nothing: both shards at their OLD heads, seq untouched.
    assert_eq!(record.seq, 0);
    assert_eq!(record.shards[0].head, anchor_zero);
    assert_eq!(record.shards[1].head, anchor_three);
    let reopened = open_branch("wal", &refs, &registry)?;
    assert_eq!(reopened.shard_current_root(0), Some(anchor_zero));
    assert_eq!(reopened.shard_current_root(3), Some(anchor_three));
    assert_eq!(
        read_at(&store, anchor_zero, b"zero")?,
        Some(b"base".to_vec())
    );
    assert_eq!(read_at(&store, anchor_zero, b"zero-new")?, None);
    // Orphans only: the half-written shard-0 head survives on disk,
    // referenced by nothing — leak-safe, reused verbatim by any retry.
    assert!(store.get(&orphan)?.is_some());
    Ok(())
}

#[test]
fn crash_at_the_barrier_leaves_old_record_valid() -> Result<(), TestError> {
    // Kill at step 3: every node of the new head is on disk (data synced,
    // directory entries unfenced), but no record names it — old record
    // valid, orphans only.
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));
    let orphan = expected_head(BASE, &[(b"key", b"value")])?;

    let anchor = {
        let mut disk = DiskStore::new(&store_dir)?;
        let anchor = build_tree(&mut disk, BASE)?;
        disk.sync_dirty_dirs()?;
        let mut refs = BranchRefStore::open(&refs_dir)?;
        let registry = BranchRegistry::new();
        let branch = create_branch("wal", [(0, anchor)], &mut refs, &registry, 10)?;
        branch.put(0, b"key", b"value")?;
        let mut store = FailpointStore {
            inner: disk,
            puts_allowed: Cell::new(None),
            fail_barrier: true,
        };
        let result = commit_branch(
            &branch,
            &mut store,
            &registry,
            durable(&mut refs, 20),
            TreePolicy::V1_DEFAULT,
        );
        assert!(matches!(result, Err(BranchCommitError::Barrier(_))));
        anchor
    };

    let store = DiskStore::new(&store_dir)?;
    let refs = BranchRefStore::open(&refs_dir)?;
    let registry = BranchRegistry::new();
    let record = refs.get("wal").ok_or(TestError::Missing("wal record"))?;
    assert_eq!(record.seq, 0);
    assert_eq!(record.shards[0].head, anchor);
    let reopened = open_branch("wal", &refs, &registry)?;
    assert_eq!(reopened.current_root(), anchor);
    assert_eq!(read_at(&store, anchor, b"key")?, None);
    assert!(
        store.get(&orphan)?.is_some(),
        "orphan nodes survive, unreferenced"
    );
    Ok(())
}

#[test]
fn crash_between_barrier_and_install_leaves_old_record_valid() -> Result<(), TestError> {
    // Kill in the step 3 → step 4 window: the barrier ran, the rename did
    // not. On disk that is "fully durable nodes, old record" — simulated
    // exactly by a volatile commit plus an explicit barrier, then the kill.
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));

    let (anchor, would_be_head) = {
        let mut store = DiskStore::new(&store_dir)?;
        let anchor = build_tree(&mut store, BASE)?;
        store.sync_dirty_dirs()?;
        let mut refs = BranchRefStore::open(&refs_dir)?;
        let registry = BranchRegistry::new();
        let branch = create_branch("wal", [(0, anchor)], &mut refs, &registry, 10)?;
        branch.put(0, b"key", b"value")?;
        commit_branch(
            &branch,
            &mut store,
            &registry,
            CommitRequest {
                durability: CommitDurability::Volatile,
                extra_parents: &[],
                timestamp: 20,
            },
            TreePolicy::V1_DEFAULT,
        )?;
        store.sync_dirty_dirs()?;
        (anchor, branch.current_root())
    };

    let store = DiskStore::new(&store_dir)?;
    let refs = BranchRefStore::open(&refs_dir)?;
    let registry = BranchRegistry::new();
    let record = refs.get("wal").ok_or(TestError::Missing("wal record"))?;
    assert_eq!(record.seq, 0, "no record advanced: the rename never ran");
    let reopened = open_branch("wal", &refs, &registry)?;
    assert_eq!(reopened.current_root(), anchor);
    // The fully materialised head is an orphan: present, unreferenced.
    assert_ne!(would_be_head, anchor);
    assert!(store.get(&would_be_head)?.is_some());
    assert_eq!(read_at(&store, anchor, b"key")?, None);
    Ok(())
}