haematite 0.5.0

//! §13 crash-window tests, POST-rename half (LEDGER A1 stage 4).
//!
//! Companions to `crash_window_tests.rs` (the pre-rename windows), over the
//! same abandon-everything kill simulation from `crash_support.rs`: once the
//! step-4 rename has happened, recovery lands on the NEW heads via
//! `open_branch` no matter where the process died afterwards (§5 steps 5-6
//! are memory-only); the rename's atomicity is pinned by the temp-file
//! survival simulation; and a multi-shard durable commit recovers every
//! shard's new head together (§7's all-or-nothing, success side — the crash
//! side is `crash_window_tests::crash_mid_materialisation_*`).

use std::fs;

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

#[test]
fn crash_after_install_recovers_new_heads_and_skips_volatile_intermediates() -> Result<(), TestError>
{
    // Kill after step 4's rename: steps 5-6 are memory-only, so recovery is
    // identical whether the process died between 4 and 5 or after returning.
    // open_branch lands at the durable head; a volatile advance made AFTER
    // the durable commit is lost from the record (its nodes orphaned), per
    // the §4 contract.
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));

    let (anchor, durable_head, volatile_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"landed", b"yes")?;
        commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20))?;
        let durable_head = branch.current_root();

        branch.put(0, b"speculative", b"lost")?;
        commit_branch(
            &branch,
            &mut store,
            &registry,
            CommitRequest {
                durability: CommitDurability::Volatile,
                extra_parents: &[],
                timestamp: 30,
            },
        )?;
        (anchor, durable_head, 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, 1);
    assert_eq!(
        record.parents,
        vec![anchor],
        "parents chain durable-to-durable"
    );
    assert_eq!(record.shards[0].fork_anchor, anchor);
    let reopened = open_branch("wal", &refs, &registry)?;
    assert_eq!(reopened.current_root(), durable_head);
    assert_eq!(reopened.fork_point(), anchor);
    assert_eq!(
        read_at(&store, durable_head, b"landed")?,
        Some(b"yes".to_vec())
    );
    assert_eq!(read_at(&store, durable_head, b"speculative")?, None);
    // The post-durable volatile advance is orphaned, not resurrected.
    assert_ne!(volatile_head, durable_head);
    assert!(store.get(&volatile_head)?.is_some());
    Ok(())
}

#[test]
fn crash_mid_install_temp_file_is_swept_and_old_record_survives() -> Result<(), TestError> {
    // Rename atomicity, simulated at the byte level: a crash after
    // write_atomic's temp write + fsync but BEFORE the rename leaves a fully
    // valid temp file beside the record. Reopen must sweep it unread and
    // serve the OLD record — the §2.1 orphaned-temp disposition.
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));

    let (anchor, 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, durable(&mut refs, 20))?;
        (anchor, branch.current_root())
    };

    // Simulate the next advance dying pre-rename: its temp file holds a full
    // record image (any bytes would do — the sweep never reads it) beside
    // the installed one.
    let ref_file = sole_ref_file(&refs_dir)?;
    let stranded = refs_dir.join(".branch-crashed00.tmp");
    fs::write(&stranded, fs::read(&ref_file)?)?;

    let refs = BranchRefStore::open(&refs_dir)?;
    assert!(
        !stranded.exists(),
        "reopen must sweep the stranded temp file"
    );
    let record = refs.get("wal").ok_or(TestError::Missing("wal record"))?;
    assert_eq!(record.seq, 1);
    assert_eq!(record.shards[0].head, head);
    assert_eq!(record.shards[0].fork_anchor, anchor);
    Ok(())
}

#[test]
fn multi_shard_durable_commit_recovers_all_new_heads() -> Result<(), TestError> {
    // The success side of §7's all-or-nothing: one record, one rename, both
    // shards' new heads recovered together.
    let dir = tempfile::tempdir()?;
    let (store_dir, refs_dir) = (dir.path().join("store"), dir.path().join("refs"));

    let (heads, anchors) = {
        let mut store = DiskStore::new(&store_dir)?;
        let anchor_zero = build_tree(&mut store, &[(b"zero", b"base")])?;
        let anchor_three = build_tree(&mut store, &[(b"three", b"base")])?;
        store.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")?;
        let outcome = commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20))?;
        (outcome.heads, (anchor_zero, anchor_three))
    };

    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, 1);
    assert_eq!(record.parents, vec![anchors.0, anchors.1]);
    let reopened = open_branch("wal", &refs, &registry)?;
    for &(shard_id, head) in &heads {
        assert_eq!(reopened.shard_current_root(shard_id), Some(head));
    }
    assert_eq!(
        read_at(&store, heads[0].1, b"zero-new")?,
        Some(b"v0".to_vec())
    );
    assert_eq!(
        read_at(&store, heads[1].1, b"three-new")?,
        Some(b"v3".to_vec())
    );
    Ok(())
}