haematite 0.7.0

//! The sweep (§5): delete exactly `enumerated − marked`, per store.
//!
//! Orchestration is deliberately thin because the hard work is shared with
//! stats: [`super::stats::run`] performs ONE inventory+mark pass under the A4
//! lock and, in [`VacuumMode::Sweep`], hands back the retained complement (the
//! exact unmarked node paths, gathered per store while each inventory was live
//! — never a second unlocked enumeration). This module adds only the two
//! things the sweep owns: TOTAL refusal evaluation and the unlink loop. The
//! refusal mapping is an EXHAUSTIVE match with no wildcard arm — a new
//! `SweepBlocker` variant is a compile error in `first_refusal`, never a
//! silent pass that would let a sweep proceed past an observed blocker.
//!
//! ## Refusal totality (brief item 4, §5)
//!
//! Any blocker anywhere means ZERO deletions anywhere — the refusal is total,
//! not per-store best-effort. The hard refusals ([`VacuumError::NodeHashMismatch`],
//! `StoreWithoutWal`, `CorruptWal`, `UnresolvableRoot`, `ConfigUnreadable`,
//! `ShardBeyondCount` from record grouping, `UnexpectedLayout`) already abort
//! the shared pass before it returns, so they reach here as a propagated
//! `Err` with nothing deleted. The SOFT blockers stats merely reports
//! ([`SweepBlocker`]) are PROMOTED here to hard typed refusals via
//! [`first_refusal`], evaluated BEFORE the first unlink — so a promoted
//! blocker also deletes nothing.
//!
//! ## Unlinks are UNFENCED BY DESIGN (§5, brief write-surface law)
//!
//! The unlink loop performs no directory fsync and tolerates `NotFound` — this
//! is the existing `DeleteNode` path semantics (`store/disk` `delete_node`:
//! `remove_file`, `NotFound` → `Ok`, no parent fence). It is correct precisely
//! because a killed-mid-sweep store is a strict SUPERSET of live: a rolled-back
//! unlink (the directory entry the power loss resurrected) is simply re-deleted
//! by the next run, which recomputes the mark set from scratch and converges.
//! Fencing the unlinks would buy nothing — there is no ordering hazard a
//! rerun does not already repair — so the sweep deliberately does not pay for
//! it. The mark set is recomputed per run and NEVER persisted.
//!
//! ## Not policy, mechanism
//!
//! The complement is reachability-only: there is no age-based or
//! keep-N-versions deletion here or anywhere in the design, and no CLI knob
//! could introduce one. Where retention adjacency surfaces it is mechanism,
//! not policy — retention decisions (DATABASE-DIRECTIONS §8) remain unruled
//! and are not this tool's to make.

use std::fs;
use std::io::ErrorKind;
use std::path::PathBuf;

use super::VacuumOptions;
use super::error::VacuumError;
use super::report::{SweepBlocker, SweepSummary, VacuumMode, VacuumReport};

/// Run the sweep. `execute` false = `--dry-run` (compute + report the deletion
/// set, ZERO unlinks); true = perform the unlinks.
pub fn run(options: &VacuumOptions, execute: bool) -> Result<VacuumReport, VacuumError> {
    // One locked inventory+mark pass, with the retained complement. Hard
    // refusals abort here with nothing deleted.
    let (mut report, plan) = super::stats::run(options, VacuumMode::Sweep)?;

    // Promote every soft blocker to a total, typed refusal BEFORE any unlink —
    // refusal totality: a single blocker means zero deletions anywhere.
    if let Some(refusal) = first_refusal(&report) {
        return Err(refusal);
    }

    // Clean run: the plan IS `enumerated − marked` across every store, so its
    // length and byte weight equal the report's unmarked totals by
    // construction.
    let deleted_nodes = plan.len();
    let deleted_bytes = report.totals.unmarked_bytes;

    if execute {
        unlink_all(&plan)?;
    }

    report.sweep = Some(SweepSummary {
        executed: execute,
        deleted_nodes,
        deleted_bytes,
        deleted_paths: plan,
    });
    Ok(report)
}

/// Delete every planned node path, tolerating `NotFound` for idempotence
/// (existing `DeleteNode` semantics, §5). A genuine I/O failure propagates —
/// a partial deletion is safe (the store stays a superset of live; a rerun
/// converges), so the error names the path and stops rather than pretending.
fn unlink_all(plan: &[PathBuf]) -> Result<(), VacuumError> {
    for path in plan {
        match fs::remove_file(path) {
            Ok(()) => {}
            Err(error) if error.kind() == ErrorKind::NotFound => {}
            Err(error) => {
                return Err(VacuumError::Io {
                    path: path.clone(),
                    error,
                });
            }
        }
    }
    Ok(())
}

/// Map the first (highest-precedence) soft blocker the pass observed to its
/// typed refusal, wiring the already-declared `VacuumError` variants — never
/// inventing one. Precedence matters only when several coexist (each §11
/// refusal pin constructs exactly one shape); the order follows the design's
/// state-first, trust-root-first bias:
///
/// 1. the platform durability floor (nothing can sweep here at all);
/// 2. store-model wrongness (our model of a directory is wrong);
/// 3. the metadata trust root (unattested / lying manifest);
/// 4. state-first: a non-terminal manifest entry;
/// 5. a missing, unreadable, or typo'd metadata source.
///
/// The mapping is split into [`precedence_rank`] and [`blocker_refusal`], each
/// an EXHAUSTIVE match with NO wildcard arm — so a future `SweepBlocker`
/// variant is a COMPILE ERROR here, never a silent `None` that would let the
/// sweep proceed past an observed blocker (refusal totality cannot fail open).
fn first_refusal(report: &VacuumReport) -> Option<VacuumError> {
    // `min_by_key` returns the FIRST element of an equal-rank tie, so within a
    // rank the earliest-observed blocker wins deterministically.
    report
        .sweep_blockers
        .iter()
        .min_by_key(|blocker| precedence_rank(blocker))
        .map(|blocker| blocker_refusal(report, blocker))
}

/// Refusal precedence — lower refuses first. EXHAUSTIVE by construction: no
/// `_` arm, so adding a `SweepBlocker` variant fails to compile until its rank
/// is stated here.
const fn precedence_rank(blocker: &SweepBlocker) -> u8 {
    match blocker {
        // 1. platform durability floor.
        SweepBlocker::UnsupportedDurability => 0,
        // 2. store-model wrongness.
        SweepBlocker::ShardBeyondCount { .. } | SweepBlocker::MalformedStoreEntries { .. } => 1,
        // 3. metadata trust root.
        SweepBlocker::MetadataUnattested | SweepBlocker::StaleManifest { .. } => 2,
        // 4. state-first: a non-terminal manifest entry.
        SweepBlocker::ManifestEntryUnsettled { .. } => 3,
        // 5. missing / unreadable / typo'd source.
        SweepBlocker::ManifestSourceMissing { .. }
        | SweepBlocker::SourceUnreadable { .. }
        | SweepBlocker::SuppliedPathMissing { .. } => 4,
    }
}

/// Map one observed blocker to its typed refusal, wiring the already-declared
/// `VacuumError` variants — never inventing one. EXHAUSTIVE by construction:
/// no `_` arm, so a new `SweepBlocker` variant must be wired here to compile.
fn blocker_refusal(report: &VacuumReport, blocker: &SweepBlocker) -> VacuumError {
    match blocker {
        SweepBlocker::UnsupportedDurability => VacuumError::UnsupportedDurability,
        SweepBlocker::ShardBeyondCount { id, shard_count } => VacuumError::ShardBeyondCount {
            id: *id,
            shard_count: *shard_count,
        },
        SweepBlocker::MalformedStoreEntries { shard_id, .. } => {
            malformed_refusal(report, *shard_id)
        }
        SweepBlocker::MetadataUnattested => VacuumError::MetadataUnattested {
            unlisted_canonical: None,
        },
        SweepBlocker::StaleManifest { unlisted } => VacuumError::MetadataUnattested {
            unlisted_canonical: Some(unlisted.clone()),
        },
        SweepBlocker::ManifestEntryUnsettled { path, state } => {
            VacuumError::MetadataSourceUnsettled {
                path: path.clone(),
                state: state.clone(),
            }
        }
        SweepBlocker::ManifestSourceMissing { path }
        | SweepBlocker::SourceUnreadable { path, .. }
        | SweepBlocker::SuppliedPathMissing { path } => {
            VacuumError::MetadataPathMissing { path: path.clone() }
        }
    }
}

/// Reconstruct the `UnexpectedLayout` refusal for a store whose enumeration
/// found malformed entries (§4: deleting by complement under a wrong model
/// deletes evidence), naming the first malformed path the report captured.
fn malformed_refusal(report: &VacuumReport, shard_id: usize) -> VacuumError {
    let store = report
        .data_dir
        .join(format!("shard-{shard_id}"))
        .join("store");
    let (path, reason) = report
        .shards
        .iter()
        .find(|shard| shard.shard_id == shard_id)
        .and_then(|shard| shard.malformed.first())
        .map_or_else(
            || (store.clone(), "malformed store entry".to_owned()),
            |entry| (entry.path.clone(), entry.reason.clone()),
        );
    VacuumError::UnexpectedLayout {
        store,
        path,
        reason,
    }
}