haematite 0.7.0

//! The verified mark walk (§2, §4 ⟨r2, B2⟩).
//!
//! This is the vacuum's caller of the shared reachability engine
//! (`crate::reachability`, itself the reusable form of
//! `branch::prune::collect_reachable`), extended the two ways the design
//! requires:
//!
//! - COMPRESSED file bytes are accounted (the operator-visible number), not
//!   prune's logical serialised bytes;
//! - 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.
//!
//! The dual fail-mode the design names lives here, in how each entry point
//! maps [`ReachError::Missing`]: [`walk_declared`] (M1/M2, manifest-bound M3)
//! turns it into a typed `UnresolvableRoot` refusal — a declared root MUST
//! resolve where its source says it lives; [`probe_store`] (unbound M3) turns
//! it into [`ProbeOutcome::Missing`] so the caller can try the next store.

use std::cell::Cell;
use std::collections::HashSet;
use std::fs;
use std::path::Path;

use crate::reachability::{NodeSource, ReachError, collect_reachable};
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 = std::collections::HashMap<Hash, u64>;

/// The vacuum's reachability source for one store: a verified file map.
///
/// Divergences from prune's store source, all required by §2/§4:
/// - resolves from the pre-enumerated [`StoreInventory`] file map, not a live
///   `NodeStore::get`;
/// - RECOMPUTES the content hash of every node it reads before trusting its
///   children (⟨r2, B2⟩);
/// - reports COMPRESSED file bytes as the byte weight;
/// - short-circuits any hash already in `marks` (a prior root in THIS store
///   already resolved its whole subtree) without re-reading or re-verifying —
///   preserving both the mark dedup across roots and the exact
///   hash-verification count.
struct VerifiedFileSource<'a> {
    inventory: &'a StoreInventory,
    store_dir: &'a Path,
    marks: &'a StoreMarks,
    verified: Cell<usize>,
}

impl NodeSource for VerifiedFileSource<'_> {
    type Error = VacuumError;
    type Bytes = u64;

    fn resolve(&self, hash: Hash, stack: &mut Vec<Hash>) -> Result<Option<u64>, VacuumError> {
        if let Some(&bytes) = self.marks.get(&hash) {
            // Already marked by a completed walk in this store: its whole
            // subtree resolved here, so resolve WITHOUT re-reading or
            // re-verifying and push no children.
            return Ok(Some(bytes));
        }
        let Some(file) = self.inventory.nodes.get(&hash) else {
            return Ok(None);
        };
        let node = read_verified(self.store_dir, file, hash)?;
        self.verified.set(self.verified.get() + 1);
        push_children(&node, stack);
        Ok(Some(file.compressed_len))
    }
}

/// 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 reached = {
        let src = VerifiedFileSource {
            inventory,
            store_dir,
            marks,
            verified: Cell::new(0),
        };
        let result = collect_reachable(&src, [root]);
        *verified += src.verified.get();
        match result {
            Ok(reached) => reached,
            Err(ReachError::Source(error)) => return Err(error),
            Err(ReachError::Missing(missing)) => {
                return Err(VacuumError::UnresolvableRoot {
                    source: source.clone(),
                    root,
                    missing,
                });
            }
        }
    };
    for (hash, bytes) in reached {
        marks.insert(hash, bytes);
    }
    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 src = VerifiedFileSource {
        inventory,
        store_dir,
        marks,
        verified: Cell::new(0),
    };
    let result = collect_reachable(&src, [root]);
    *verified += src.verified.get();
    match result {
        Ok(reached) => Ok(ProbeOutcome::Resolved(reached.into_keys().collect())),
        Err(ReachError::Source(error)) => Err(error),
        Err(ReachError::Missing(missing)) => Ok(ProbeOutcome::Missing(missing)),
    }
}

/// 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)
}