liminal/durability/

error.rs

//! Named storage failures and uncertain channel append outcomes.
/// Error taxonomy for haematite-backed durability operations.
#[derive(Debug, thiserror::Error)]
pub enum DurabilityError {
    /// A logical operation has no staged rows to commit.
    #[error("operation batch is empty")]
    EmptyOperationBatch,

    /// A store does not implement atomic routed storage.
    #[error("{store} does not support atomic routed storage")]
    BatchUnsupported {
        /// The store's type.
        store: &'static str,
    },

    /// The store did not establish whether a sent append committed.
    #[error("append to {stream_key} at sequence {sequence} is unresolved: {cause}")]
    AppendUnresolved {
        /// Partition stream receiving the append.
        stream_key: String,
        /// Reserved sequence of the sent bytes.
        sequence: u64,
        /// Original append failure, never used to classify the outcome.
        cause: String,
    },

    /// The authoritative read of an earlier uncertain append failed.
    #[error(
        "append to {stream_key} at sequence {sequence} remains unresolved: append: {append_error}; read-back: {read_error}"
    )]
    PendingAppendUnreadable {
        /// Partition stream that must be reconciled.
        stream_key: String,
        /// Reserved sequence of the sent bytes.
        sequence: u64,
        /// Original append error, retained across every failed read.
        append_error: String,
        /// The latest authoritative read error.
        read_error: String,
    },

    /// A write the store had started when its caller stopped waiting: it may
    /// have been applied. It is never treated as not written: the caller
    /// reads the position back to settle it (`settle`), or refuses by this
    /// name.
    #[error(
        "write outcome unknown on shard {pid}: it may have been applied; read the position again"
    )]
    OutcomeUnknown {
        /// The shard that had started the write.
        pid: u64,
    },

    /// A write withdrawn before the store started it: nothing was written,
    /// and it may be sent again.
    #[error("write not started on shard {pid}: nothing was written")]
    NotStarted {
        /// The shard the write was queued for.
        pid: u64,
    },

    /// Haematite returned a store-level failure.
    ///
    /// The umbrella `From<haematite::ApiError>` conversion lives in
    /// [`super::store`] because it routes the optimistic-concurrency variants to
    /// their dedicated cases rather than wrapping them here.
    #[error("haematite store error: {0}")]
    StoreError(haematite::ApiError),

    /// An append observed a different stream sequence than the caller expected.
    #[error("sequence conflict: expected {expected}, actual {actual}")]
    SequenceConflict {
        /// Caller-provided expected stream sequence.
        expected: u64,
        /// Actual stream sequence reported by haematite.
        actual: u64,
    },

    /// A cursor checkpoint attempted to move from a stale stored value.
    #[error("cursor regression: stored {stored}, attempted {attempted}")]
    CursorRegression {
        /// Current value stored for the cursor.
        stored: u64,
        /// Value the caller attempted to checkpoint from.
        attempted: u64,
    },

    /// A producer idempotency key collided with an existing dedup entry.
    #[error("dedup key collision for key {key}")]
    DedupCollision {
        /// Idempotency key that collided.
        key: String,
    },

    /// Durability configuration failed validation.
    #[error("configuration error: {0}")]
    ConfigError(String),

    /// An ephemeral durable store could not be opened on disk.
    ///
    /// Raised only on the construction path of a self-owned ephemeral store
    /// (see [`super::EphemeralRoot::open_store`]); the guarding temporary directory
    /// is already removed by the time this surfaces.
    #[error("ephemeral store open failed: {0}")]
    EphemeralStoreOpen(String),

    /// An operation reached an ephemeral store whose backing database was
    /// already detached by teardown.
    ///
    /// Unreachable by construction: the store is detached only inside the
    /// ephemeral guard's `Drop`, which cannot overlap a live handle's call.
    /// The variant exists because that detachment is expressed through an
    /// `Option` the compiler cannot see through, and the fallback must be a
    /// typed refusal rather than a panic.
    #[error("ephemeral store already detached by teardown")]
    EphemeralStoreDetached,

    /// A store that cannot delete entries was asked to truncate a stream.
    #[error("{store} cannot truncate {stream_key} below sequence {before_seq}")]
    TruncateUnsupported {
        /// The store's type.
        store: &'static str,
        /// The stream it was asked to cut.
        stream_key: String,
        /// The sequence it was asked to cut below.
        before_seq: u64,
    },

    /// A store that cannot list a routed key window was asked to scan one.
    #[error("{store} cannot scan routed keys on route {route_key}")]
    RoutedScanUnsupported {
        /// The store's type.
        store: &'static str,
        /// The route it was asked to scan.
        route_key: String,
    },

    /// A store that cannot delete a routed key window in one commit was
    /// asked to remove routed rows.
    ///
    /// Composing the window from single deletes is never done in its place
    /// (DUR-018 R3): each would be its own commit.
    #[error("{store} cannot remove routed rows on route {route_key} in one commit")]
    RoutedRemoveUnsupported {
        /// The store's type.
        store: &'static str,
        /// The route it was asked to cut.
        route_key: String,
    },

    /// A store that cannot delete a key inside a routed batch was asked to
    /// commit a batch with a delete (DUR-028 R3). Nothing was written: the
    /// batch is one durable operation or none.
    #[error("{store} cannot delete inside a routed batch on route {route_key}")]
    RoutedDeleteUnsupported {
        /// The store's type.
        store: &'static str,
        /// The route the batch was for.
        route_key: String,
    },

    /// Persisted envelope bytes could not be encoded or decoded.
    #[error("envelope serialization error: {0}")]
    EnvelopeError(String),
}

impl From<haematite::SequenceConflict> for DurabilityError {
    fn from(error: haematite::SequenceConflict) -> Self {
        Self::SequenceConflict {
            expected: error.expected,
            actual: error.actual,
        }
    }
}

impl From<haematite::CasMismatch> for DurabilityError {
    fn from(error: haematite::CasMismatch) -> Self {
        // The real `CasMismatch` carries `Option<u64>` to distinguish absent
        // (`None`) from stored-zero (`Some(0)`). The cursor contract treats an
        // absent key as the value 0 (see `HaematiteStore::cas`), so a `None`
        // collapses to 0 here when reporting a regression.
        Self::CursorRegression {
            stored: error.actual.unwrap_or(0),
            attempted: error.expected.unwrap_or(0),
        }
    }
}