haematite 0.7.0

//! Locating and rebuilding the affected leaf window (CHUNKING-POLICY.md §2.1).
//!
//! A mutation touches a contiguous run of leaves. The window's edges must rest
//! on boundaries that CANNOT change when the window's contents change, or the
//! rebuilt subtree would not be canonical.
//!
//! Under v1 (key-only boundaries, no `boundary_before`) the left edge is always
//! stable: the leaf holding the smallest mutated key begins right after a
//! previous leaf that closed on a stable `boundary_after` key. So v1 needs no
//! left extension, and the right edge extends only while the last entry is open.
//!
//! Under v2 both edges need care. `boundary_before` (oversized isolation) is
//! NOT stable — deleting or shrinking the oversized entry that created it merges
//! leftward — so the window must extend LEFT to the nearest stable
//! `boundary_after` edge, and the right edge stops either on a `boundary_after`
//! OR before the next leaf's stable `boundary_before` (an oversized singleton).

use crate::store::NodeStore;

use super::super::cursor::{TreeError, load_node};
use super::super::node::Node;
use super::super::policy::TreePolicy;
use super::{ChildRef, Entry, Mutation};

/// Half-open `[start, end)` range of leaf indices the mutation may change.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct Window {
    pub(super) start: usize,
    pub(super) end: usize,
}

/// Determine the contiguous run of existing leaves whose entries can change,
/// from the mutation key span alone. The left edge here is provisional: under
/// v2 [`rebuild_window`] extends it further left to the nearest stable
/// `boundary_after`. Rightward spillover is handled during the rebuild.
pub(super) fn affected_window(
    leaves: &[ChildRef],
    mutations: &[Mutation],
) -> Result<Option<Window>, TreeError> {
    let Some(min_key) = mutations.iter().map(|m| m.key.as_slice()).min() else {
        return Ok(None);
    };
    let Some(max_key) = mutations.iter().map(|m| m.key.as_slice()).max() else {
        return Ok(None);
    };

    if leaves.is_empty() {
        // Only possible for a corrupt tree; the empty tree always has one leaf.
        return Err(TreeError::InvalidNode);
    }

    let start = leaf_index_for_key(leaves, min_key);
    let end = leaf_index_for_key(leaves, max_key)
        .saturating_add(1)
        .min(leaves.len());

    Ok(Some(Window { start, end }))
}

/// Index of the leaf whose key range contains `key` (the last leaf whose
/// separator is `<= key`, or leaf 0 if `key` precedes all separators).
fn leaf_index_for_key(leaves: &[ChildRef], key: &[u8]) -> usize {
    let mut index = 0;
    for (position, (separator, _hash)) in leaves.iter().enumerate() {
        if separator.as_slice() <= key {
            index = position;
        } else {
            break;
        }
    }
    index
}

/// Re-derive the leaves of the affected window from their entries (with
/// mutations applied) using the policy's content-defined boundaries.
///
/// Under v2 the window is first extended LEFT while the preceding leaf's last
/// entry is not a stable `boundary_after` edge (so an unstable `boundary_before`
/// left edge cannot leave a stale leaf behind — the B1 fix). The right edge then
/// extends while the last rebuilt entry is open AND the next leaf does not begin
/// with a stable `boundary_before` singleton.
pub(super) fn rebuild_window<S: NodeStore + ?Sized>(
    store: &mut S,
    leaves: &[ChildRef],
    window: Window,
    mutations: &[Mutation],
    policy: TreePolicy,
) -> Result<Vec<ChildRef>, TreeError> {
    let start = extend_left(store, leaves, window.start, policy)?;

    let mut entries = load_entries(store, &leaves[start..window.end])?;
    apply_mutations(&mut entries, mutations)?;

    // Right edge: pull in successive leaves until the last surviving entry rests
    // on a stable boundary. Stop before a next leaf that begins with an oversized
    // (`boundary_before`) entry — that edge is already stable.
    let mut consumed_end = window.end;
    while consumed_end < leaves.len() && last_entry_is_open(&entries, policy) {
        let next = load_entries(store, std::slice::from_ref(&leaves[consumed_end]))?;
        if policy.is_v2() && next_leaf_is_isolated(next.first(), policy) {
            break;
        }
        entries.extend(next);
        consumed_end = consumed_end.saturating_add(1);
    }

    let new_leaves = super::spine::store_leaf_replacements(store, entries, policy)?;

    let mut result =
        Vec::with_capacity(start + new_leaves.len() + leaves.len().saturating_sub(consumed_end));
    result.extend_from_slice(&leaves[..start]);
    result.extend(new_leaves);
    result.extend_from_slice(&leaves[consumed_end..]);
    Ok(result)
}

/// v2-only left extension: walk left while a preceding leaf exists and its last
/// entry is not a stable `boundary_after` edge. v1 left edges are always stable,
/// so this returns `start` unchanged there (the preceding leaf always closed on
/// a `boundary_after` key) — gated on `is_v2` to avoid the extra leaf loads.
fn extend_left<S: NodeStore + ?Sized>(
    store: &S,
    leaves: &[ChildRef],
    start: usize,
    policy: TreePolicy,
) -> Result<usize, TreeError> {
    if !policy.is_v2() {
        return Ok(start);
    }
    let mut start = start;
    while start > 0 {
        let prev = load_entries(store, std::slice::from_ref(&leaves[start - 1]))?;
        let Some((key, value)) = prev.last() else {
            // A stored leaf is never empty except the sole empty-tree leaf, which
            // has no predecessor; treat an unexpected empty leaf as a stable edge.
            break;
        };
        if policy.leaf_boundary_after(key.as_slice(), value.len()) {
            break;
        }
        start -= 1;
    }
    Ok(start)
}

/// True when the last entry does NOT close its chunk, so a re-chunk would leave
/// an unterminated trailing chunk that must absorb the next leaf.
fn last_entry_is_open(entries: &[Entry], policy: TreePolicy) -> bool {
    entries
        .last()
        .is_some_and(|(key, value)| !policy.leaf_boundary_after(key.as_slice(), value.len()))
}

/// True when the next leaf begins with an oversized (`boundary_before`) entry,
/// i.e. it is a stable isolated singleton the window must not swallow.
fn next_leaf_is_isolated(first: Option<&Entry>, policy: TreePolicy) -> bool {
    first.is_some_and(|(key, value)| policy.leaf_boundary_before(key.len(), value.len()))
}

fn load_entries<S: NodeStore + ?Sized>(
    store: &S,
    leaves: &[ChildRef],
) -> Result<Vec<Entry>, TreeError> {
    let mut entries = Vec::new();
    for (_separator, hash) in leaves {
        match &*load_node(store, *hash)? {
            Node::Leaf(leaf) => entries.extend_from_slice(leaf.entries()),
            Node::Internal(_internal) => return Err(TreeError::InvalidNode),
        }
    }
    Ok(entries)
}

fn apply_mutations(entries: &mut Vec<Entry>, mutations: &[Mutation]) -> Result<(), TreeError> {
    for mutation in mutations {
        let search =
            entries.binary_search_by(|(key, _value)| key.as_slice().cmp(mutation.key.as_slice()));
        match (&mutation.value, search) {
            (Some(value), Ok(index)) => {
                let Some((_key, stored_value)) = entries.get_mut(index) else {
                    return Err(TreeError::InvalidNode);
                };
                value.clone_into(stored_value);
            }
            (Some(value), Err(index)) => {
                if index > entries.len() {
                    return Err(TreeError::InvalidNode);
                }
                entries.insert(index, (mutation.key.clone(), value.clone()));
            }
            (None, Ok(index)) => {
                if index >= entries.len() {
                    return Err(TreeError::InvalidNode);
                }
                entries.remove(index);
            }
            (None, Err(_index)) => {}
        }
    }

    Ok(())
}