haematite 0.6.1

//! §10 history-independence property test at the COMMIT layer (LEDGER A1
//! stage 4; brief §10, pinned spec).
//!
//! The tree layer's `mutate_history_independence_tests.rs` proves root =
//! f(final key→value set) for `batch_mutate`. This module proves the commit
//! path adds no bytes of its own on top of that: a random final map, reached
//! through a random op sequence with interleaved overwrites and deletes,
//! partitioned into a random number of commit batches, yields per-shard head
//! roots identical to a single-batch reference commit — across `Volatile` and
//! `Durable` modes and 1/2/3-shard handles. The one way the commit path could
//! break this is by writing commit metadata (seq counters, branch names,
//! parent links) into the tree keyspace — §8's hard design rule forbids it,
//! and this property is the tripwire. Consequence for urd R13: identical
//! landed history ⇒ identical roots ⇒ deterministic projection input.

use std::collections::BTreeMap;

use proptest::prelude::*;
use proptest::test_runner::TestCaseError;

use super::commit::{CommitDurability, CommitRequest, commit_branch};
use super::fork::fork_shards_registered;
use super::handle::{BranchHandle, ShardId};
use super::lifecycle::create_branch;
use super::refstore::BranchRefStore;
use super::registry::BranchRegistry;
use crate::store::MemoryStore;
use crate::tree::{Hash, LeafNode, Node};

/// A buffered branch operation: `Some(value)` = put, `None` = delete.
type Op = (Vec<u8>, Option<Vec<u8>>);

/// The shard layouts the property runs over: 1, 2 and 3 shards, with
/// deliberately non-contiguous ids so nothing accidentally assumes density.
const SHARD_SETS: [&[ShardId]; 3] = [&[0], &[0, 2], &[0, 2, 5]];

/// Whether every batch commits volatile or durable (§4 — both must land on
/// the same head roots; durability changes what survives, never the hash).
#[derive(Debug, Clone, Copy)]
enum Mode {
    Volatile,
    Durable,
}

/// Wraps any displayable failure into a proptest case failure, keeping the
/// helpers unwrap-free (house law: no unwrap/expect, even in tests).
fn fail(context: &str, error: impl std::fmt::Display) -> TestCaseError {
    TestCaseError::fail(format!("{context}: {error}"))
}

/// The canonical empty root (an empty leaf), matching the commit path.
fn empty_root(store: &mut MemoryStore) -> Result<Hash, TestCaseError> {
    let leaf = LeafNode::new(Vec::new()).map_err(|error| fail("empty leaf", error))?;
    Ok(store.put(&Node::Leaf(leaf)))
}

/// Deterministic splitmix-style generator for shuffles and batch cuts.
struct Lcg(u64);

impl Lcg {
    const fn next_u64(&mut self) -> u64 {
        self.0 = self
            .0
            .wrapping_mul(0x5851_f42d_4c95_7f2d)
            .wrapping_add(0x1405_7b7e_f767_814f);
        self.0 >> 16
    }
}

/// Fisher-Yates shuffle driven by [`Lcg`].
fn shuffle<T: Clone>(items: &[T], lcg: &mut Lcg) -> Vec<T> {
    let mut shuffled = items.to_vec();
    for index in (1..shuffled.len()).rev() {
        let pick = usize::try_from(lcg.next_u64()).unwrap_or(usize::MAX) % (index + 1);
        shuffled.swap(index, pick);
    }
    shuffled
}

/// Build a put/delete sequence whose final map is exactly `final_pairs`:
/// the final pairs in a seed-dependent order, braided with noise puts for
/// keys outside the final map, wrong-value overwrites that are corrected in
/// place, and trailing deletes of every noise key.
fn build_ops(
    final_pairs: &[(Vec<u8>, Vec<u8>)],
    noise: &[(Vec<u8>, Vec<u8>)],
    seed: u64,
) -> Vec<Op> {
    let final_map: BTreeMap<&[u8], &[u8]> = final_pairs
        .iter()
        .map(|(key, value)| (key.as_slice(), value.as_slice()))
        .collect();
    let noise: Vec<&(Vec<u8>, Vec<u8>)> = noise
        .iter()
        .filter(|(key, _value)| !final_map.contains_key(key.as_slice()))
        .collect();

    let mut lcg = Lcg(seed.wrapping_add(0x9e37_79b9_7f4a_7c15));
    let order = shuffle(final_pairs, &mut lcg);

    let mut ops: Vec<Op> = Vec::new();
    let (mut final_index, mut noise_index) = (0usize, 0usize);
    while final_index < order.len() || noise_index < noise.len() {
        let pick_final = lcg.next_u64() & 1 == 0;
        if (pick_final && final_index < order.len()) || noise_index >= noise.len() {
            let (key, value) = &order[final_index];
            ops.push((key.clone(), Some(value.clone())));
            final_index += 1;
        } else {
            let (key, value) = noise[noise_index];
            ops.push((key.clone(), Some(value.clone())));
            noise_index += 1;
        }
    }
    for (index, (key, value)) in order.iter().enumerate() {
        if index % 3 == 0 {
            let mut wrong = value.clone();
            wrong.push(0xff);
            ops.push((key.clone(), Some(wrong)));
            ops.push((key.clone(), Some(value.clone())));
        }
    }
    for (key, _value) in &noise {
        ops.push((key.clone(), None));
    }
    ops
}

/// Seed-dependent batch boundaries: after op `i` for every `i` in the cut
/// set, `run_commits` commits the buffered prefix.
fn pick_cuts(op_count: usize, cut_count: usize, seed: u64) -> Vec<usize> {
    if op_count == 0 {
        return Vec::new();
    }
    let mut lcg = Lcg(seed ^ 0xdead_beef_cafe_f00d);
    let mut cuts: Vec<usize> = (0..cut_count)
        .map(|_| usize::try_from(lcg.next_u64()).unwrap_or(usize::MAX) % op_count)
        .collect();
    cuts.sort_unstable();
    cuts.dedup();
    cuts
}

/// Route a key to one of the handle's shards by its first byte, so every
/// multi-shard case genuinely spreads work.
fn route(key: &[u8], shards: &[ShardId]) -> ShardId {
    let byte = key.first().copied().unwrap_or(0);
    shards[usize::from(byte) % shards.len()]
}

/// Independent replay of the op stream into a plain map, for the sanity
/// check that the generated history really converges on `final_pairs`.
fn replay(ops: &[Op]) -> BTreeMap<Vec<u8>, Vec<u8>> {
    let mut map = BTreeMap::new();
    for (key, value) in ops {
        match value {
            Some(value) => {
                map.insert(key.clone(), value.clone());
            }
            None => {
                map.remove(key);
            }
        }
    }
    map
}

/// Apply `ops` to a fresh branch over `store`, committing at every cut and
/// once at the end, and return the final per-shard heads (ascending id).
fn run_commits(
    store: &mut MemoryStore,
    shards: &[ShardId],
    ops: &[Op],
    cuts: &[usize],
    mode: Mode,
) -> Result<Vec<(ShardId, Hash)>, TestCaseError> {
    let registry = BranchRegistry::new();
    let anchor = empty_root(store)?;
    let roots: Vec<(ShardId, Hash)> = shards.iter().map(|&shard| (shard, anchor)).collect();

    // Durable mode owns a real on-disk ref store for the case's lifetime;
    // volatile mode is an anonymous registered fork.
    let durable_refs = match mode {
        Mode::Volatile => None,
        Mode::Durable => {
            let dir = tempfile::tempdir().map_err(|error| fail("tempdir", error))?;
            let refs =
                BranchRefStore::open(dir.path()).map_err(|error| fail("refs open", error))?;
            Some((dir, refs))
        }
    };
    let (branch, mut refs) = match durable_refs {
        None => {
            let handle =
                fork_shards_registered(roots, &registry).map_err(|error| fail("fork", error))?;
            (handle, None)
        }
        Some((dir, mut refs)) => {
            let handle = create_branch("hi", roots, &mut refs, &registry, 1)
                .map_err(|error| fail("create", error))?;
            (handle, Some((dir, refs)))
        }
    };

    let mut timestamp = 1u64;
    let commit = |branch: &BranchHandle,
                  store: &mut MemoryStore,
                  refs: Option<&mut BranchRefStore>,
                  timestamp: &mut u64|
     -> Result<(), TestCaseError> {
        *timestamp += 1;
        let durability = refs.map_or(CommitDurability::Volatile, |refs| {
            CommitDurability::Durable { refs }
        });
        commit_branch(
            branch,
            store,
            &registry,
            CommitRequest {
                durability,
                extra_parents: &[],
                timestamp: *timestamp,
            },
        )
        .map(drop)
        .map_err(|error| fail("commit", error))
    };

    for (index, (key, value)) in ops.iter().enumerate() {
        let shard = route(key, shards);
        value
            .as_ref()
            .map_or_else(
                || branch.delete(shard, key),
                |value| branch.put(shard, key, value),
            )
            .map_err(|error| fail("buffer", error))?;
        if cuts.binary_search(&index).is_ok() {
            commit(
                &branch,
                store,
                refs.as_mut().map(|(_dir, refs)| refs),
                &mut timestamp,
            )?;
        }
    }
    commit(
        &branch,
        store,
        refs.as_mut().map(|(_dir, refs)| refs),
        &mut timestamp,
    )?;

    let mut heads = Vec::with_capacity(shards.len());
    for &shard in shards {
        let head = branch
            .shard_current_root(shard)
            .ok_or_else(|| fail("head", format!("missing shard {shard}")))?;
        heads.push((shard, head));
    }

    // Durable extra: every durable commit installed a record, so the on-disk
    // heads must equal the in-memory heads shard for shard.
    if let Some((_dir, refs)) = &refs {
        let record = refs
            .get("hi")
            .ok_or_else(|| fail("record", "branch record vanished"))?;
        let recorded: Vec<(ShardId, Hash)> = record
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .collect();
        if recorded != heads {
            return Err(fail(
                "record heads",
                format!("record {recorded:?} != in-memory {heads:?}"),
            ));
        }
    }
    Ok(heads)
}

/// Distinct random keys (1-5 bytes) with random values (0-5 bytes).
fn key_value_set(count: std::ops::Range<usize>) -> impl Strategy<Value = Vec<(Vec<u8>, Vec<u8>)>> {
    proptest::collection::btree_map(
        proptest::collection::vec(any::<u8>(), 1..6),
        proptest::collection::vec(any::<u8>(), 0..6),
        count,
    )
    .prop_map(|map| map.into_iter().collect())
}

proptest! {
    #![proptest_config(ProptestConfig {
        cases: 64,
        max_shrink_iters: 8192,
        ..ProptestConfig::default()
    })]

    /// §10 pinned property: any partition of any converging op history into
    /// commit batches — volatile or durable, 1/2/3 shards — lands every shard
    /// on the head root of the single-batch reference build.
    #[test]
    fn prop_commit_batching_never_changes_the_head_roots(
        shard_index in 0usize..SHARD_SETS.len(),
        durable in any::<bool>(),
        final_pairs in key_value_set(1..48),
        noise in key_value_set(0..32),
        seed in any::<u64>(),
        cut_count in 0usize..6,
    ) {
        let shards = SHARD_SETS[shard_index];
        let mode = if durable { Mode::Durable } else { Mode::Volatile };
        let ops = build_ops(&final_pairs, &noise, seed);
        let cuts = pick_cuts(ops.len(), cut_count, seed);

        // Sanity: the generated history really converges on the final map.
        let target: BTreeMap<Vec<u8>, Vec<u8>> = final_pairs.iter().cloned().collect();
        prop_assert_eq!(&replay(&ops), &target, "generated ops did not reach the final map");

        // Reference: the final map applied as ONE batch, committed once,
        // volatile — the pure f(final key→value set) build.
        let reference_ops: Vec<Op> = final_pairs
            .iter()
            .map(|(key, value)| (key.clone(), Some(value.clone())))
            .collect();

        // One shared store for both builds: content addressing makes the
        // roots store-independent, and sharing exercises the dedup reuse the
        // §4 soundness note leans on.
        let mut store = MemoryStore::new();
        let batched = run_commits(&mut store, shards, &ops, &cuts, mode)?;
        let reference = run_commits(&mut store, shards, &reference_ops, &[], Mode::Volatile)?;

        prop_assert_eq!(
            batched,
            reference,
            "commit batching changed a head root (mode={:?}, shards={:?}, ops={}, cuts={:?})",
            mode,
            shards,
            ops.len(),
            cuts
        );
    }
}