haematite 0.7.0

//! Stream-and-rebuild: re-materialise a tree's ENTIRE logical map from empty
//! under a target [`TreePolicy`], returning the new canonical root.
//!
//! The canonical operation behind two CHUNKING-POLICY.md §4.2 crossings: the
//! per-shard migration rebuild (step 3) and the v1-anchor FORK crossing (step 5,
//! `fork_at` under a v2 database). Reads are POLICY-FREE — the walk routes by
//! ordered separators, so a tree of ANY chunking streams identically; only the
//! rebuild re-chunks. HI canonicity (`branch/hi_proptest.rs`) makes this an
//! IDENTITY when the source is already in `policy`'s shape: every rebuilt node
//! dedupes on write, so re-crossing an already-target-shaped root is a
//! structural no-op producing the byte-identical root.

use crate::store::NodeStore;

use super::cursor::TreeError;
use super::mutate::batch_mutate_owned;
use super::node::{Hash, LeafNode, Node};
use super::policy::TreePolicy;

/// Rebuild `source_root`'s entire logical map from empty under `policy`.
///
/// Streams every `(key, value)` entry (RAW — the TTL/stamp envelope is copied
/// verbatim, never decoded) and applies them as ONE `batch_mutate_owned` from an
/// empty leaf, so the result is the fresh-build canonical root of the same map
/// under `policy`.
pub fn stream_and_rebuild<S: NodeStore + ?Sized>(
    store: &mut S,
    source_root: Hash,
    policy: TreePolicy,
) -> Result<Hash, TreeError> {
    let mut entries = Vec::new();
    collect_leaf_entries(store, source_root, &mut entries)?;
    let empty = store
        .put(&Node::Leaf(LeafNode::new(Vec::new())?))
        .map_err(|_| TreeError::InvalidNode)?;
    if entries.is_empty() {
        return Ok(empty);
    }
    let batch: Vec<(Vec<u8>, Option<Vec<u8>>)> = entries
        .into_iter()
        .map(|(key, value)| (key, Some(value)))
        .collect();
    batch_mutate_owned(store, empty, batch, policy)
}

/// Collect every leaf entry reachable from `root` (order-independent — the
/// rebuild's `batch_mutate_owned` normalises/sorts the batch). An iterative
/// stack walk so a tall tree never overflows the native stack.
fn collect_leaf_entries<S: NodeStore + ?Sized>(
    store: &S,
    root: Hash,
    out: &mut Vec<(Vec<u8>, Vec<u8>)>,
) -> Result<(), TreeError> {
    let mut stack = vec![root];
    while let Some(hash) = stack.pop() {
        let node = store
            .get(&hash)
            .map_err(|_| TreeError::InvalidNode)?
            .ok_or(TreeError::MissingNode { hash })?;
        match &*node {
            Node::Leaf(leaf) => {
                for (key, value) in leaf.entries() {
                    out.push((key.clone(), value.clone()));
                }
            }
            Node::Internal(internal) => {
                for (_separator, child) in internal.children() {
                    stack.push(*child);
                }
            }
        }
    }
    Ok(())
}