haematite 0.7.0

//! Typed vacuum refusals (STORAGE-VACUUM.md §7).
//!
//! House error law, plus the ⟨r15, A2⟩ remedy-naming law: every refusal with
//! a known innocent cause names its REMEDY in the error text — a refusal that
//! names its recovery is fail-closed; one that doesn't is fail-stuck.

use std::fmt;
use std::io;
use std::path::{Path, PathBuf};

use crate::tree::Hash;

/// Where a mark root came from, for refusals that must name their source.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MarkSourceId {
    /// M1: a shard WAL's committed-root marker.
    WalRoot { shard_id: usize },
    /// M2: an HBR1 branch record's fork anchor for one shard.
    BranchAnchor { branch: String, shard_id: usize },
    /// M2: an HBR1 branch record's committed head for one shard.
    BranchHead { branch: String, shard_id: usize },
    /// M3: a named snapshot's root hash from an HSR1 registry.
    Snapshot { name: String, registry: PathBuf },
}

impl fmt::Display for MarkSourceId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::WalRoot { shard_id } => {
                write!(formatter, "shard-{shard_id} WAL committed root")
            }
            Self::BranchAnchor { branch, shard_id } => {
                write!(
                    formatter,
                    "branch {branch:?} fork anchor (shard-{shard_id})"
                )
            }
            Self::BranchHead { branch, shard_id } => {
                write!(formatter, "branch {branch:?} head (shard-{shard_id})")
            }
            Self::Snapshot { name, registry } => {
                write!(formatter, "snapshot {name:?} ({})", registry.display())
            }
        }
    }
}

/// Typed vacuum refusals. Every variant names what the operator must decide.
#[derive(Debug)]
pub enum VacuumError {
    /// Another live writer holds the A4 data-dir lock; the vacuum refuses
    /// loudly and never queues (§6).
    Locked {
        lock_path: PathBuf,
        /// ⟨r4, M5⟩ the losing acquisition may still have CREATED the
        /// anchor (create-then-lose-the-race); disclosure survives here.
        anchor_created: bool,
    },
    /// The advisory lockfile could not be opened or locked for an I/O reason
    /// distinct from contention.
    LockIo {
        lock_path: PathBuf,
        error: io::Error,
        /// Whether the failed acquisition CREATED the anchor before the lock
        /// syscall failed — the ⟨r4, M5⟩ transparency rule holds on error
        /// paths too: the sole write exception is named whenever it occurs.
        anchor_created: bool,
    },
    /// ⟨r15, C1⟩ `config.json` is absent or unreadable. There is no shard
    /// iteration without a trusted `shard_count`, and guessing one would make
    /// the M1 mark source silently incomplete under the very tool that
    /// promises a complete inventory. Refuses BOTH stats and sweep.
    ConfigUnreadable { path: PathBuf, reason: String },
    /// §2 M1: a `shard-{id}` directory exists without a decodable WAL. The
    /// WAL is created eagerly at boot (`DurableWal::new` is create-on-open
    /// with a parent fence, `wal/durable.rs`), so this is an observable
    /// partial or damaged state — never laziness. No sweep anywhere.
    StoreWithoutWal {
        shard_id: usize,
        shard_dir: PathBuf,
        /// Whether `store/` exists alongside the missing WAL — present-store
        /// means possible forensic damage; absent-store is the innocent
        /// crash-between-dir-and-boot transient.
        store_present: bool,
    },
    /// §2 M1: a shard WAL is present but CRC-corrupt or undecodable. Fail
    /// closed, consistent with the recovery-hardening design.
    CorruptWal { shard_id: usize, detail: String },
    /// ⟨r2, B2⟩ a canonically-named node file's bytes hash differently than
    /// the name claims. The file is corruption evidence and is preserved;
    /// the store is never swept around it.
    NodeHashMismatch { store: PathBuf, path: PathBuf },
    /// A mark root (or one of its descendants) failed to fully resolve in
    /// the store(s) its source declares (§2 M2/M3). A marked root that
    /// cannot fully resolve is corruption evidence.
    UnresolvableRoot {
        source: MarkSourceId,
        root: Hash,
        missing: Hash,
    },
    /// §4: a directory's layout does not match the model. Deleting by
    /// complement under a wrong model deletes evidence or data.
    UnexpectedLayout {
        store: PathBuf,
        path: PathBuf,
        reason: String,
    },
    /// ⟨r15, C1⟩ / §2 M2: a shard id at or beyond the config's
    /// `shard_count` — evidence the config's model of this directory is
    /// wrong, or the data belongs to a different database.
    ShardBeyondCount { id: usize, shard_count: usize },
    /// §3 ⟨r2, B1⟩: sweep requires the metadata universe positively
    /// established (manifest or attestation); when a manifest exists, a
    /// canonical source it omits is proof the manifest lies.
    MetadataUnattested {
        /// The unlisted canonical source, in the stale-manifest case ⟨r4⟩.
        unlisted_canonical: Option<PathBuf>,
    },
    /// ⟨r4, B4⟩ an explicitly supplied metadata path does not exist, or a
    /// settled `Published` entry's source is absent/unreadable ⟨r8, M7⟩.
    MetadataPathMissing { path: PathBuf },
    /// ⟨r7, B9⟩ a manifest entry in a non-terminal state (`Reserved`,
    /// `Publishing`, `Updating`, `Decommissioning`) — regardless of the
    /// path's filesystem condition. The durable state is the only honest
    /// signal.
    MetadataSourceUnsettled { path: PathBuf, state: String },
    /// ⟨r9⟩⟨r10⟩⟨r11⟩ no real directory-entry durability barrier exists on
    /// this platform/filesystem; deleting operations refuse rather than
    /// proceed on a no-op fence. Only `vacuum_stats` is universal.
    UnsupportedDurability,
    /// A metadata source exists but could not be opened or decoded — a
    /// dropped record is a dropped pin, so decode fails loud (§2 M2).
    MetadataOpen { path: PathBuf, reason: String },
    /// Non-refusal I/O failure while reading the directory tree. Named
    /// escape hatch for raw I/O errors that carry no policy meaning.
    Io { path: PathBuf, error: io::Error },
}

impl fmt::Display for VacuumError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Locked {
                lock_path,
                anchor_created,
            } => {
                fmt_locked(formatter, lock_path)?;
                if *anchor_created {
                    formatter.write_str(
                        "; note: this losing run CREATED the empty advisory writer.lock \
                         anchor before the race was lost — it is inert and safe to leave",
                    )?;
                }
                Ok(())
            }
            Self::LockIo {
                lock_path,
                error,
                anchor_created,
            } => {
                write!(
                    formatter,
                    "failed to open or lock {}: {error}",
                    lock_path.display()
                )?;
                if *anchor_created {
                    formatter.write_str(
                        "; note: this failed run CREATED the empty advisory writer.lock \
                         anchor before the lock syscall failed — it is inert (the OS lock \
                         state is the only truth; an unlocked anchor means nothing) and \
                         safe to leave in place",
                    )?;
                }
                Ok(())
            }
            Self::ConfigUnreadable { path, reason } => {
                fmt_config_unreadable(formatter, path, reason)
            }
            Self::StoreWithoutWal {
                shard_id,
                shard_dir,
                store_present,
            } => fmt_store_without_wal(formatter, *shard_id, shard_dir, *store_present),
            Self::CorruptWal { shard_id, detail } => write!(
                formatter,
                "shard-{shard_id} WAL is present but undecodable ({detail}); a corrupt WAL \
                 is a forensic incident — the vacuum fails closed rather than marking from \
                 a partial root set"
            ),
            Self::NodeHashMismatch { store, path } => fmt_hash_mismatch(formatter, store, path),
            Self::UnresolvableRoot {
                source,
                root,
                missing,
            } => fmt_unresolvable(formatter, source, *root, *missing),
            Self::UnexpectedLayout {
                store,
                path,
                reason,
            } => fmt_unexpected_layout(formatter, store, path, reason),
            Self::ShardBeyondCount { id, shard_count } => write!(
                formatter,
                "shard id {id} is at or beyond this database's shard_count {shard_count}; \
                 either the config's model of this directory is wrong or the data belongs \
                 to a different database — resolve which before any sweep"
            ),
            Self::MetadataUnattested { unlisted_canonical } => {
                fmt_unattested(formatter, unlisted_canonical.as_deref())
            }
            Self::MetadataPathMissing { path } => fmt_path_missing(formatter, path),
            Self::MetadataSourceUnsettled { path, state } => fmt_unsettled(formatter, path, state),
            Self::UnsupportedDurability => formatter.write_str(
                "this platform/filesystem provides no real directory-entry durability \
                 barrier; deleting operations refuse rather than proceed on a no-op fence \
                 (supported contract: Unix over local filesystems) — vacuum_stats remains \
                 available everywhere",
            ),
            Self::MetadataOpen { path, reason } => fmt_metadata_open(formatter, path, reason),
            Self::Io { path, error } => {
                write!(formatter, "I/O error at {}: {error}", path.display())
            }
        }
    }
}

fn fmt_locked(formatter: &mut fmt::Formatter<'_>, lock_path: &Path) -> fmt::Result {
    write!(
        formatter,
        "data dir writer lock at {} is held by a live writer; the vacuum never runs beside \
         one and never queues — stop the database process that owns this directory, then \
         re-run",
        lock_path.display()
    )
}

fn fmt_config_unreadable(
    formatter: &mut fmt::Formatter<'_>,
    path: &Path,
    reason: &str,
) -> fmt::Result {
    write!(
        formatter,
        "config.json at {} is absent or unreadable ({reason}); there is no shard iteration \
         without a trusted shard_count and the vacuum will not guess one — verify this is \
         the database's data_dir and restore its config.json, then re-run",
        path.display()
    )
}

fn fmt_store_without_wal(
    formatter: &mut fmt::Formatter<'_>,
    shard_id: usize,
    shard_dir: &Path,
    store_present: bool,
) -> fmt::Result {
    write!(
        formatter,
        "shard-{shard_id} at {} exists without a decodable shard.wal; every successfully \
         booted shard has one (the WAL is created eagerly at first boot). ",
        shard_dir.display()
    )?;
    if store_present {
        formatter.write_str(
            "A store exists alongside the missing WAL: treat this as forensic evidence of \
             damage — the vacuum will not convert a lost WAL into node loss",
        )
    } else {
        formatter.write_str(
            "This is the shape of a crash between shard-directory creation and first boot: \
             reopen the database once to re-materialise the shard (recreating its WAL), \
             then re-run the vacuum",
        )
    }
}

fn fmt_hash_mismatch(formatter: &mut fmt::Formatter<'_>, store: &Path, path: &Path) -> fmt::Result {
    write!(
        formatter,
        "node file {} in store {} does not hash to the name it is stored under; the file \
         is corruption evidence and has been preserved — no part of this store will be \
         swept until an operator investigates",
        path.display(),
        store.display()
    )
}

fn fmt_unresolvable(
    formatter: &mut fmt::Formatter<'_>,
    source: &MarkSourceId,
    root: Hash,
    missing: Hash,
) -> fmt::Result {
    write!(
        formatter,
        "mark root {root} from {source} does not fully resolve in its declared store(s): \
         node {missing} is missing or unreadable; a marked root that cannot fully resolve \
         is corruption evidence and the vacuum refuses rather than sweeping around it"
    )
}

fn fmt_unexpected_layout(
    formatter: &mut fmt::Formatter<'_>,
    store: &Path,
    path: &Path,
    reason: &str,
) -> fmt::Result {
    write!(
        formatter,
        "unexpected layout at {} (store {}): {reason}; the directory does not match the \
         vacuum's model, and deleting by complement under a wrong model deletes evidence \
         or data",
        path.display(),
        store.display()
    )
}

fn fmt_unattested(formatter: &mut fmt::Formatter<'_>, unlisted: Option<&Path>) -> fmt::Result {
    match unlisted {
        Some(path) => write!(
            formatter,
            "the metadata manifest omits the canonical source at {}, which exists on disk — \
             the manifest lies and no sweep will trust it; repair the manifest through the \
             haem manifest operations before sweeping",
            path.display()
        ),
        None => formatter.write_str(
            "no metadata manifest exists and attest_metadata_complete is false; a report \
             cannot reveal an omitted path, so sweep refuses until the metadata universe is \
             positively established — supply a manifest, or attest that the canonical plus \
             supplied paths are ALL metadata for this data dir",
        ),
    }
}

fn fmt_path_missing(formatter: &mut fmt::Formatter<'_>, path: &Path) -> fmt::Result {
    write!(
        formatter,
        "metadata path {} does not exist; attestation attests completeness, not existence — \
         verify the path (a typo here would silently omit the real directory) or remove it \
         from the invocation, then re-run",
        path.display()
    )
}

fn fmt_unsettled(formatter: &mut fmt::Formatter<'_>, path: &Path, state: &str) -> fmt::Result {
    write!(
        formatter,
        "manifest entry for {} is in non-terminal state {state}; filesystem condition \
         proves nothing here and sweep requires every listed entry settled — run the \
         matching haem manifest operation (reconcile for Publishing/Updating, \
         remove-reservation for Reserved, resume or force-remove for Decommissioning), \
         then re-run",
        path.display()
    )
}

fn fmt_metadata_open(formatter: &mut fmt::Formatter<'_>, path: &Path, reason: &str) -> fmt::Result {
    write!(
        formatter,
        "metadata source {} exists but could not be decoded: {reason}; a dropped record is \
         a dropped pin, so the vacuum fails loud rather than proceeding with a partial \
         mark set",
        path.display()
    )
}

impl std::error::Error for VacuumError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::LockIo { error, .. } | Self::Io { error, .. } => Some(error),
            Self::Locked { .. }
            | Self::ConfigUnreadable { .. }
            | Self::StoreWithoutWal { .. }
            | Self::CorruptWal { .. }
            | Self::NodeHashMismatch { .. }
            | Self::UnresolvableRoot { .. }
            | Self::UnexpectedLayout { .. }
            | Self::ShardBeyondCount { .. }
            | Self::MetadataUnattested { .. }
            | Self::MetadataPathMissing { .. }
            | Self::MetadataSourceUnsettled { .. }
            | Self::UnsupportedDurability
            | Self::MetadataOpen { .. } => None,
        }
    }
}