haematite 0.6.2

//! The verified mark walk (§2, §4 ⟨r2, B2⟩).
//!
//! `branch::prune::collect_reachable`'s dedup/traversal/fail-loud semantics,
//! extended two ways the design requires: compressed file bytes are
//! accounted (the operator-visible number), and every traversed node's hash
//! is RECOMPUTED from its canonical serialised bytes before that node's
//! children are trusted — syntactic path validity proves nothing about
//! contents, and a misplaced copy under a valid name would let the walk
//! traverse the WRONG child list and sweep real descendants.

use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::Path;

use crate::store::disk::decompress_node;
use crate::tree::{Hash, Node};

use super::error::{MarkSourceId, VacuumError};
use super::inventory::{NodeFile, StoreInventory};

/// Marks committed against one shard store: hash → compressed file bytes.
pub type StoreMarks = HashMap<Hash, u64>;

/// Walk `root` where its source DECLARES it lives (§2 M1/M2, manifest-bound
/// M3): every node must resolve in `inventory`; any missing or unreadable
/// referenced node is a typed refusal — the vacuum reports and refuses
/// rather than sweeping around corruption evidence.
pub fn walk_declared(
    inventory: &StoreInventory,
    store_dir: &Path,
    root: Hash,
    source: &MarkSourceId,
    marks: &mut StoreMarks,
    verified: &mut usize,
) -> Result<(), VacuumError> {
    let mut stack = vec![root];
    while let Some(hash) = stack.pop() {
        if marks.contains_key(&hash) {
            continue;
        }
        let Some(file) = inventory.nodes.get(&hash) else {
            return Err(VacuumError::UnresolvableRoot {
                source: source.clone(),
                root,
                missing: hash,
            });
        };
        let node = read_verified(store_dir, file, hash)?;
        *verified += 1;
        push_children(&node, &mut stack);
        marks.insert(hash, file.compressed_len);
    }
    Ok(())
}

/// A probe's outcome: the resolved subtree, or the first node found absent.
#[derive(Debug)]
pub enum ProbeOutcome {
    Resolved(HashSet<Hash>),
    Missing(Hash),
}

/// Probe whether `root` FULLY resolves in `inventory` (§2 M3, unbound
/// registries): returns the resolved subtree's hashes, or the first hash
/// absent from this store (which is not an error for a probe — the root may
/// resolve elsewhere). Hash mismatches are NEVER tolerated, probe or not: a
/// mismatched file is corruption evidence wherever it is found.
pub fn probe_store(
    inventory: &StoreInventory,
    store_dir: &Path,
    root: Hash,
    marks: &StoreMarks,
    verified: &mut usize,
) -> Result<ProbeOutcome, VacuumError> {
    let mut resolved = HashSet::new();
    let mut stack = vec![root];
    while let Some(hash) = stack.pop() {
        if resolved.contains(&hash) {
            continue;
        }
        if marks.contains_key(&hash) {
            // Already marked in this store by a completed walk, so its whole
            // subtree resolved here; nothing further to prove.
            resolved.insert(hash);
            continue;
        }
        let Some(file) = inventory.nodes.get(&hash) else {
            return Ok(ProbeOutcome::Missing(hash));
        };
        let node = read_verified(store_dir, file, hash)?;
        *verified += 1;
        push_children(&node, &mut stack);
        resolved.insert(hash);
    }
    Ok(ProbeOutcome::Resolved(resolved))
}

/// Commit a probe's resolved set into the store's marks.
pub fn commit_marks(inventory: &StoreInventory, resolved: &HashSet<Hash>, marks: &mut StoreMarks) {
    for hash in resolved {
        if let Some(file) = inventory.nodes.get(hash) {
            marks.insert(*hash, file.compressed_len);
        }
    }
}

fn push_children(node: &Node, stack: &mut Vec<Hash>) {
    if let Node::Internal(internal) = node {
        stack.extend(
            internal
                .children()
                .iter()
                .map(|(_lower_bound, child_hash)| *child_hash),
        );
    }
}

/// Read one node file and require its contents to be EXACTLY the canonical
/// serialisation of a node hashing to `expected` (⟨r2, B2⟩). Any failure to
/// decompress, deserialise, re-serialise byte-identically, or hash-match is
/// the same finding: the file is not what its name claims.
fn read_verified(store_dir: &Path, file: &NodeFile, expected: Hash) -> Result<Node, VacuumError> {
    let mismatch = || VacuumError::NodeHashMismatch {
        store: store_dir.to_path_buf(),
        path: file.path.clone(),
    };
    let compressed = fs::read(&file.path).map_err(|error| VacuumError::Io {
        path: file.path.clone(),
        error,
    })?;
    let serialised = decompress_node(&compressed).map_err(|_error| mismatch())?;
    let node = Node::deserialise(&serialised).map_err(|_error| mismatch())?;
    let (canonical, computed) = node.serialise_and_hash();
    if computed != expected || canonical != serialised {
        return Err(mismatch());
    }
    Ok(node)
}