holt 0.9.2

//! Top-level error type.

use crate::layout::BlobGuid;
use crate::store::{AllocError, FreeError};

/// Result alias used throughout the crate.
pub type Result<T, E = Error> = std::result::Result<T, E>;

/// Whether a failed [`crate::DB::atomic`] call may have applied its batch.
///
/// The classification describes only the batch passed to that call. Holt may
/// flush writes acknowledged by earlier calls before it starts the new batch.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum AtomicErrorKind {
    /// The failed batch did not apply any of its mutations.
    ///
    /// The caller may retry it, subject to normal guard revalidation and any
    /// concurrent changes made after the failed attempt.
    DefinitelyNotApplied,
    /// Holt crossed the batch mutation boundary before the failure.
    ///
    /// The batch may be fully applied, partly applied in the live process, or
    /// absent after recovery. The caller must not retry it blindly.
    OutcomeUnknown,
}

pub(crate) fn is_blob_store_not_found(error: &Error) -> bool {
    matches!(
        error,
        Error::BlobStoreIo(err) if err.kind() == std::io::ErrorKind::NotFound
    )
}

/// Top-level error type covering the union of every failure mode.
///
/// Marked `#[non_exhaustive]` — new variants may be added in
/// minor releases without breaking SemVer. Callers should always
/// match with a `_` arm.
#[derive(Debug)]
#[non_exhaustive]
pub enum Error {
    /// BlobStore I/O failure.
    BlobStoreIo(std::io::Error),
    /// Bump allocator / slot table exhaustion / invalid alloc.
    Alloc(AllocError),
    /// Free-list misuse.
    Free(FreeError),
    /// Key longer than `u16::MAX` bytes.
    KeyTooLong {
        /// Caller-supplied length.
        len: usize,
    },
    /// Value longer than `u16::MAX` bytes.
    ValueTooLong {
        /// Caller-supplied length.
        len: usize,
    },
    /// A walker arm hit a code path that the engine doesn't yet
    /// implement (e.g. degenerate `Leaf` / `EmptyRoot` spillover
    /// or strict-prefix ART insert cases). The static string names
    /// the unimplemented case for diagnostics.
    NotYetImplemented(&'static str),
    /// An internal invariant the engine relies on was observed
    /// to be violated — typically a background thread closing a
    /// channel it shouldn't have closed, or a completion sender
    /// disappearing without producing a result. Distinct from
    /// [`Self::NotYetImplemented`] (genuine feature gap) and
    /// from [`Self::NodeCorrupt`] (on-disk / cache layout
    /// problem). The static string names the specific invariant
    /// for triage.
    Internal(&'static str),
    /// A classified failure from [`crate::DB::atomic`].
    ///
    /// `kind` states whether the requested batch may have applied. `source`
    /// retains the underlying validation, storage, walker, or journal error.
    Atomic {
        /// Commit-boundary classification for the failed batch.
        kind: AtomicErrorKind,
        /// Underlying Holt error.
        source: Box<Error>,
    },
    /// A blob's slot table or header is corrupt — recovery
    /// should bail out rather than silently misbehave.
    ///
    /// Construct via [`Error::node_corrupt`] and optionally
    /// enrich with [`Error::with_blob_guid`] / [`Error::with_slot`]
    /// when the surrounding code path knows the affected blob
    /// or slot. The buffer manager and walker entry points
    /// automatically attach blob context where they have it.
    NodeCorrupt {
        /// Static description of where the corruption was detected.
        context: &'static str,
        /// GUID of the blob that exposed the corruption, when
        /// the propagating code path knows it. `None` for low-
        /// level helpers that don't have blob context (e.g. raw
        /// `BlobFrame::wrap` on a stack buffer).
        blob_guid: Option<BlobGuid>,
        /// Slot inside `blob_guid` that exposed the corruption,
        /// when applicable. `None` for header / body-level
        /// problems.
        slot: Option<u16>,
    },
    /// WAL replay encountered a record whose sanity validation
    /// failed — record magic mismatch, CRC32 mismatch, unknown
    /// variant tag, truncated body, etc.
    ReplaySanityFailed {
        /// What went wrong (decoder-supplied static string).
        context: &'static str,
        /// Position in the journal file where the bad record
        /// starts. `0` when the codec is invoked on a raw
        /// in-memory buffer and the caller hasn't supplied an
        /// offset.
        record_offset: u64,
    },
    /// An attached journal operation requires a supported DB profile whose
    /// stream has been initialized with an application genesis anchor.
    JournalStreamUnavailable {
        /// Static reason the stream cannot serve the request.
        reason: &'static str,
    },
    /// The caller's attached-journal cursor is older than the local
    /// checkpoint floor and its envelopes are no longer retained.
    JournalPositionExpired {
        /// Sequence requested by the caller.
        requested: u64,
        /// Oldest sequence retained by the local journal.
        checkpoint: u64,
    },
    /// The caller supplied an anchor that conflicts with Holt's anchor at the
    /// same sequence or does not name the current stream tail.
    JournalAnchorMismatch {
        /// Sequence supplied by the caller.
        requested: u64,
        /// Sequence Holt expected at this boundary.
        expected: u64,
    },
    /// One encoded WAL record cannot fit in Holt's bounded submit buffer.
    WalRecordTooLarge {
        /// Encoded record size requested by the caller.
        bytes: usize,
        /// Maximum encoded record size accepted by this Holt build.
        maximum: usize,
    },
    /// An attached-journal page request used an invalid limit.
    InvalidJournalScanLimit {
        /// Static reason the limit is invalid.
        reason: &'static str,
    },
    /// `Tree::rename` (or similar) called with a `src` that has no
    /// leaf in the tree.
    NotFound,
    /// `Tree::rename(.., force=false)` called with a `dst` that
    /// already has a leaf. Caller can retry with `force=true` to
    /// overwrite.
    DstExists,
    /// A named DB tree was requested but no catalog entry exists.
    TreeNotFound {
        /// Requested tree name.
        name: String,
    },
    /// A named DB tree create was requested for an existing tree.
    TreeExists {
        /// Existing tree name.
        name: String,
    },
    /// A named DB tree name is invalid.
    InvalidTreeName {
        /// Static reason.
        reason: &'static str,
    },
    /// A named DB tree handle was used after `DB::drop_tree`.
    TreeDropped,
    /// A mutation or maintenance operation used a read-only handle.
    ReadOnly,
    /// A scoped [`crate::View`] read tried to access a key or range
    /// prefix outside the subtree captured when the view was opened.
    OutsideViewScope {
        /// Length of the requested key or prefix.
        requested_len: usize,
        /// Length of the view's captured prefix.
        scope_len: usize,
    },
    /// [`crate::Tree::gc`] was called on a tree opened through a `DB`.
    /// DB trees share one buffer manager, so reclaiming unreachable
    /// frames safely needs a DB-wide pass rather than a single tree's.
    GcRequiresStandaloneTree,
    /// The monotonic copy-on-write snapshot epoch reached its largest
    /// durable value. Existing data remains valid, but no new snapshot can
    /// be registered without wrapping the counter.
    SnapshotEpochExhausted,
}

impl Error {
    /// Construct a [`Error::NodeCorrupt`] with the given static
    /// context but no blob / slot metadata. Layers higher in the
    /// stack (buffer manager, walker entry points) typically
    /// enrich the error via [`Self::with_blob_guid`] /
    /// [`Self::with_slot`] before it surfaces to the caller.
    #[must_use]
    pub const fn node_corrupt(context: &'static str) -> Self {
        Self::NodeCorrupt {
            context,
            blob_guid: None,
            slot: None,
        }
    }

    /// Attach a `blob_guid` to a [`Self::NodeCorrupt`] error
    /// without overwriting one set by a deeper layer. No-op for
    /// other variants — safe to chain unconditionally on any
    /// `Error` value.
    #[must_use]
    pub fn with_blob_guid(mut self, guid: BlobGuid) -> Self {
        if let Self::NodeCorrupt { blob_guid, .. } = &mut self {
            if blob_guid.is_none() {
                *blob_guid = Some(guid);
            }
        }
        self
    }

    /// Attach a `slot` index to a [`Self::NodeCorrupt`] error
    /// without overwriting one set by a deeper layer.
    #[must_use]
    pub fn with_slot(mut self, slot_index: u16) -> Self {
        if let Self::NodeCorrupt { slot, .. } = &mut self {
            if slot.is_none() {
                *slot = Some(slot_index);
            }
        }
        self
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BlobStoreIo(e) => write!(f, "store I/O: {e}"),
            Self::Alloc(e) => write!(f, "alloc: {e}"),
            Self::Free(e) => write!(f, "free: {e}"),
            Self::KeyTooLong { len } => write!(f, "key too long ({len} bytes; max {})", u16::MAX),
            Self::ValueTooLong { len } => {
                write!(f, "value too long ({len} bytes; max {})", u16::MAX)
            }
            Self::NotYetImplemented(where_) => write!(f, "not yet implemented: {where_}"),
            Self::Internal(what) => write!(f, "internal invariant violated: {what}"),
            Self::Atomic { kind, source } => match kind {
                AtomicErrorKind::DefinitelyNotApplied => {
                    write!(f, "atomic batch definitely not applied: {source}")
                }
                AtomicErrorKind::OutcomeUnknown => {
                    write!(f, "atomic batch outcome unknown: {source}")
                }
            },
            Self::NodeCorrupt {
                context,
                blob_guid,
                slot,
            } => {
                write!(f, "node corrupt at {context}")?;
                if let Some(g) = blob_guid {
                    // First 4 bytes is enough to disambiguate in
                    // logs without dumping the full 16-byte tag.
                    write!(f, " (blob={:02x?})", &g[..4])?;
                }
                if let Some(s) = slot {
                    write!(f, " (slot={s})")?;
                }
                Ok(())
            }
            Self::ReplaySanityFailed {
                context,
                record_offset,
            } => {
                write!(
                    f,
                    "WAL replay sanity-check failed at offset {record_offset}: {context}"
                )
            }
            Self::JournalStreamUnavailable { reason } => {
                write!(f, "attached journal stream unavailable: {reason}")
            }
            Self::JournalPositionExpired {
                requested,
                checkpoint,
            } => write!(
                f,
                "attached journal position {requested} expired at checkpoint {checkpoint}"
            ),
            Self::JournalAnchorMismatch {
                requested,
                expected,
            } => {
                if requested == expected {
                    write!(f, "attached journal digest mismatch at position {requested}")
                } else {
                    write!(
                        f,
                        "attached journal anchor {requested} does not match expected position {expected}"
                    )
                }
            }
            Self::WalRecordTooLarge { bytes, maximum } => write!(
                f,
                "WAL record is too large ({bytes} bytes; maximum {maximum})"
            ),
            Self::InvalidJournalScanLimit { reason } => {
                write!(f, "invalid attached journal scan limit: {reason}")
            }
            Self::NotFound => write!(f, "key not found"),
            Self::DstExists => write!(
                f,
                "destination key already exists (use force=true to overwrite)"
            ),
            Self::TreeNotFound { name } => write!(f, "DB tree not found: {name}"),
            Self::TreeExists { name } => write!(f, "DB tree already exists: {name}"),
            Self::InvalidTreeName { reason } => write!(f, "invalid DB tree name: {reason}"),
            Self::TreeDropped => write!(f, "DB tree has been dropped"),
            Self::ReadOnly => write!(f, "tree is read-only"),
            Self::OutsideViewScope {
                requested_len,
                scope_len,
            } => write!(
                f,
                "view access outside captured scope (requested {requested_len} bytes, scope {scope_len} bytes)"
            ),
            Self::GcRequiresStandaloneTree => write!(
                f,
                "gc is only supported on standalone trees; trees opened through a DB share a buffer manager"
            ),
            Self::SnapshotEpochExhausted => write!(f, "snapshot epoch exhausted"),
        }
    }
}

impl std::error::Error for Error {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::BlobStoreIo(e) => Some(e),
            Self::Alloc(e) => Some(e),
            Self::Free(e) => Some(e),
            Self::Atomic { source, .. } => Some(source.as_ref()),
            _ => None,
        }
    }
}

impl From<std::io::Error> for Error {
    fn from(e: std::io::Error) -> Self {
        Self::BlobStoreIo(e)
    }
}
impl From<AllocError> for Error {
    fn from(e: AllocError) -> Self {
        Self::Alloc(e)
    }
}
impl From<FreeError> for Error {
    fn from(e: FreeError) -> Self {
        Self::Free(e)
    }
}

#[cfg(test)]
mod tests {
    use std::error::Error as StdError;
    use std::io;

    use super::*;

    #[test]
    fn node_corrupt_context_is_enriched_once() {
        let err = Error::node_corrupt("slot decode")
            .with_blob_guid([0xAB; 16])
            .with_blob_guid([0xCD; 16])
            .with_slot(7)
            .with_slot(9);

        let rendered = err.to_string();
        assert!(rendered.contains("node corrupt at slot decode"));
        assert!(rendered.contains("blob=[ab, ab, ab, ab]"));
        assert!(rendered.contains("slot=7"));
        assert!(!rendered.contains("cd"));
        assert!(!rendered.contains("slot=9"));
    }

    #[test]
    fn display_covers_public_error_variants() {
        let cases = [
            (
                Error::KeyTooLong { len: 70_000 }.to_string(),
                "key too long (70000 bytes; max 65535)",
            ),
            (
                Error::ValueTooLong { len: 70_001 }.to_string(),
                "value too long (70001 bytes; max 65535)",
            ),
            (
                Error::NotYetImplemented("strict prefix").to_string(),
                "not yet implemented: strict prefix",
            ),
            (
                Error::Internal("lost dirty image").to_string(),
                "internal invariant violated: lost dirty image",
            ),
            (
                Error::Atomic {
                    kind: AtomicErrorKind::DefinitelyNotApplied,
                    source: Box::new(Error::TreeNotFound {
                        name: "objects".to_owned(),
                    }),
                }
                .to_string(),
                "atomic batch definitely not applied: DB tree not found: objects",
            ),
            (
                Error::Atomic {
                    kind: AtomicErrorKind::OutcomeUnknown,
                    source: Box::new(Error::Internal("journal acknowledgement lost")),
                }
                .to_string(),
                "atomic batch outcome unknown: internal invariant violated: journal acknowledgement lost",
            ),
            (
                Error::ReplaySanityFailed {
                    context: "bad CRC",
                    record_offset: 42,
                }
                .to_string(),
                "WAL replay sanity-check failed at offset 42: bad CRC",
            ),
            (Error::NotFound.to_string(), "key not found"),
            (
                Error::DstExists.to_string(),
                "destination key already exists (use force=true to overwrite)",
            ),
            (
                Error::TreeNotFound {
                    name: "objects".to_owned(),
                }
                .to_string(),
                "DB tree not found: objects",
            ),
            (
                Error::TreeExists {
                    name: "objects".to_owned(),
                }
                .to_string(),
                "DB tree already exists: objects",
            ),
            (
                Error::InvalidTreeName { reason: "empty" }.to_string(),
                "invalid DB tree name: empty",
            ),
            (Error::TreeDropped.to_string(), "DB tree has been dropped"),
            (Error::ReadOnly.to_string(), "tree is read-only"),
            (
                Error::SnapshotEpochExhausted.to_string(),
                "snapshot epoch exhausted",
            ),
        ];

        for (actual, expected) in cases {
            assert_eq!(actual, expected);
        }
    }

    #[test]
    fn display_covers_attached_journal_errors() {
        let cases = [
            (
                Error::JournalStreamUnavailable {
                    reason: "stream has not been initialized",
                }
                .to_string(),
                "attached journal stream unavailable: stream has not been initialized",
            ),
            (
                Error::JournalPositionExpired {
                    requested: 7,
                    checkpoint: 9,
                }
                .to_string(),
                "attached journal position 7 expired at checkpoint 9",
            ),
            (
                Error::JournalAnchorMismatch {
                    requested: 8,
                    expected: 9,
                }
                .to_string(),
                "attached journal anchor 8 does not match expected position 9",
            ),
            (
                Error::JournalAnchorMismatch {
                    requested: 9,
                    expected: 9,
                }
                .to_string(),
                "attached journal digest mismatch at position 9",
            ),
            (
                Error::WalRecordTooLarge {
                    bytes: 17,
                    maximum: 16,
                }
                .to_string(),
                "WAL record is too large (17 bytes; maximum 16)",
            ),
            (
                Error::InvalidJournalScanLimit {
                    reason: "row limit must be non-zero",
                }
                .to_string(),
                "invalid attached journal scan limit: row limit must be non-zero",
            ),
        ];

        for (actual, expected) in cases {
            assert_eq!(actual, expected);
        }
    }

    #[test]
    fn error_sources_are_exposed_for_wrapped_errors_only() {
        let blob = Error::from(io::Error::other("disk"));
        assert!(blob.source().is_some());

        let alloc = Error::from(AllocError::OutOfSlots);
        assert!(alloc.source().is_some());
        assert_eq!(
            alloc.to_string(),
            format!("alloc: {}", AllocError::OutOfSlots)
        );

        let free = Error::from(FreeError::InvalidSlot(99));
        assert!(free.source().is_some());
        assert_eq!(free.to_string(), "free: free_node: invalid slot index 99");

        let atomic = Error::Atomic {
            kind: AtomicErrorKind::DefinitelyNotApplied,
            source: Box::new(Error::NotFound),
        };
        assert_eq!(
            atomic.source().unwrap().to_string(),
            Error::NotFound.to_string()
        );

        assert!(Error::NotFound.source().is_none());
        assert!(Error::DstExists.source().is_none());
        assert!(Error::InvalidTreeName { reason: "empty" }
            .source()
            .is_none());
        assert!(Error::TreeDropped.source().is_none());
        assert!(Error::ReadOnly.source().is_none());
        assert!(Error::SnapshotEpochExhausted.source().is_none());
    }
}