armdb 0.9.0

use std::io;

use armour_core::persist::PersistError;

/// `" by pid 1234"` when the lock file carried a readable pid, `""` otherwise.
fn pid_hint(pid: &Option<u32>) -> String {
    match pid {
        Some(p) => format!(" by pid {p}"),
        None => String::new(),
    }
}

/// A shard that owes an unapplied committed write.
///
/// `shard` is an index into **one collection's** shard array, so `collection`
/// is what makes the pair addressable. Only Bitcask-backed collections appear
/// here: FixedStore replay reads the reserved slot's exact state, which no
/// later write can disturb, so it is never fenced.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FencedShard {
    pub collection: &'static str,
    pub shard: usize,
    pub gsn: u64,
}

/// Which step after the transaction log append failed.
///
/// Each carries a **different** post-condition, so a caller can act on the
/// stage instead of parsing the error text.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum TxFailStage {
    /// An operation failed to apply. The record stays pending and replay
    /// finishes it at the next successful open — against a **process** crash;
    /// surviving a power loss still needs `Tx::durable`. `fenced` names the
    /// shards that owe those writes; they refuse further writes until then.
    Apply,
    /// Every write applied, but releasing a participant failed. The record is
    /// settled as applied — replay has nothing left to finish — and this
    /// transaction's write-hook effects were dropped and are **not** recovered
    /// by reopening. The primary data is intact and complete.
    Finish,
    /// A `durable()` transaction whose log fsync failed after the append. The
    /// writes are applied and visible in this process, but the record is not
    /// confirmed durable: after a power loss it may be absent, and then no
    /// replay happens.
    Sync,
}

/// P2-4: `#[non_exhaustive]` so future variants are additive — downstream crates
/// must include a wildcard arm, and adding a variant (like `DuplicateCollection`)
/// is no longer a source-breaking change to their exhaustive matches.
#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum DbError {
    /// 400
    #[error("client error: {0}")]
    Client(&'static str),
    /// 404
    #[error("key not found")]
    KeyNotFound,
    /// 501
    #[error("not implemented")]
    NotImplemented,
    #[error("storage error: {0}")]
    Internal(&'static str),
    #[error(transparent)]
    Io(#[from] io::Error),
    #[error("CRC32 mismatch: expected {expected:#x}, got {actual:#x}")]
    CrcMismatch { expected: u32, actual: u32 },
    #[error("corrupted entry at offset {offset}")]
    CorruptedEntry { offset: u64 },
    #[error("invalid configuration: {0}")]
    Config(&'static str),
    #[error("CAS value mismatch")]
    CasMismatch,
    #[error("key already exists")]
    KeyExists,
    /// A copied DiskLoc points to a file that has just been removed from
    /// inner.immutable by compaction. The data is logically still live;
    /// the caller should retry with a fresh DiskLoc from the index.
    #[error("stale DiskLoc - file removed by compaction")]
    StaleDiskLoc,
    #[error("all slots are occupied")]
    SlotsFull,
    #[error("key routes to a different shard")]
    ShardMismatch,
    /// Reserved for a future multi-collection transaction API (T5).
    ///
    /// Historically produced by `Db::atomic2/3/4` when the same collection
    /// appeared in two argument positions (re-locking its shards from the same
    /// thread would self-deadlock). Those methods were removed in 0.8.0; the
    /// variant and its default 500 mapping are kept so clients and T5 stay
    /// stable.
    /// armdb does not return this variant today.
    #[error("the same collection appears twice in a cross-collection transaction")]
    DuplicateCollectionInTx,
    /// The transaction log could not be repaired after a failed append; the
    /// database must be reopened before it can accept transactions again.
    #[error("transaction log is poisoned: {0}")]
    TxLogPoisoned(String),
    /// A committed transaction operation could not be applied to this shard, so
    /// the shard owes a write that only replay can complete. It refuses further
    /// writes until the database is reopened.
    ///
    /// Bitcask replay asks whether a reserved GSN is above the shard's recovered
    /// watermark. Letting a later write land would raise that watermark past the
    /// unapplied reservation, and the next open would read the committed
    /// operation as already on disk and drop it. Reads are unaffected, other
    /// shards are unaffected, and reopening the database replays the operation
    /// and clears the fence.
    #[error(
        "shard {shard} owes a committed transaction write (gsn {gsn}); reopen the database to replay it"
    )]
    ShardFenced { shard: usize, gsn: u64 },
    /// The transaction's record was appended to the log, and the failure
    /// happened after that point. **Do not retry**: some or all of the writes
    /// may already be applied, and the record may still be replayed.
    ///
    /// What is recoverable depends on `stage` — see [`TxFailStage`]. This
    /// variant on its own promises only that the transaction is *not* a no-op.
    ///
    /// `fenced` lists the shards that owe an unapplied committed write; they
    /// refuse further writes with [`DbError::ShardFenced`] until the database is
    /// reopened. It is non-empty only for [`TxFailStage::Apply`], and only for
    /// Bitcask-backed participants.
    #[error("transaction {tx_id} failed after the log point ({stage:?}): {source}")]
    TxFailedAfterLog {
        tx_id: u64,
        stage: TxFailStage,
        fenced: Vec<FencedShard>,
        source: Box<DbError>,
    },
    /// A collection with this name is already open on this `Db`, or the name is
    /// reserved for internal state (`__seq`, `db.info`, `db.user`). Opening it
    /// again would alias two in-memory indexes onto the same shard files and
    /// silently drop each other's updates.
    #[error("collection name already in use or reserved: {0}")]
    DuplicateCollection(&'static str),
    /// The database directory is already open — by another process, or by
    /// another live handle in this one. armdb requires exclusive access: two
    /// engines over the same shard files would run independent in-memory
    /// indexes and silently drop each other's writes.
    ///
    /// `pid` is a best-effort diagnostic read from the lock file; `None` means
    /// it was unreadable, not that no process holds the lock.
    #[error("database directory is locked{}: {}", pid_hint(pid), path.display())]
    Locked {
        path: std::path::PathBuf,
        pid: Option<u32>,
    },
    /// Optimistic-concurrency retry signal: a value read in phase 1 changed before
    /// the narrow transaction in phase 2 could commit. The caller should retry.
    /// 409. armdb does not return this variant today; reserved for T5 and callers.
    #[error("transaction conflict — retry")]
    TxConflict,
    /// `db.tx().commit()` was called from inside a collection's `atomic()` closure.
    /// Shard locks are not reentrant; commit while `atomic()` holds them would hang.
    #[error("cannot commit a transaction from inside atomic()")]
    ReentrantTx,
    /// A pending transaction record's schema fingerprint does not match the
    /// collection's stored type hash at open.
    #[error(
        "pending transaction schema mismatch on '{collection}' (tx_id={tx_id}): stored={stored:#x} logged={logged:#x}"
    )]
    TxSchemaMismatch {
        collection: &'static str,
        tx_id: u64,
        stored: u64,
        logged: u64,
    },
    /// A pending FixedStore operation cannot be ordered against the slot's
    /// current version: the two are more than 2^29 apart in the 30-bit ring, so
    /// neither "already applied" nor "never applied" can be established.
    /// Recovering requires human judgement, so the open fails rather than guess.
    #[error(
        "collection {collection}: pending transaction targets slot {slot} at meta \
         {target:#010x}, which cannot be ordered against the slot's current meta {current:#010x}"
    )]
    TxFixedSlot {
        collection: &'static str,
        slot: u32,
        current: u32,
        target: u32,
    },
    #[error("projection budget exceeded: rows={rows}, bytes={bytes}")]
    ProjectionBudgetExceeded { rows: usize, bytes: usize },
    #[error("undeclared index target: {index}")]
    UndeclaredIndexTarget { index: &'static str },
    #[error("projection incomplete")]
    ProjectionIncomplete(crate::projection::PendingProjection),
    /// Two source owners projected the same `(index, key)` during
    /// [`crate::schema::repair`]. Carries the same detail as the matching
    /// `Finding::IndexKeyCollision` so the offending rows can be found without
    /// a second `validate` pass; each owner is `(source collection, hex key)`.
    #[error("index key collision during repair on `{index}`: key {key} owned by {owners:?}")]
    IndexKeyCollision {
        index: &'static str,
        key: String,
        owners: Vec<(String, String)>,
    },
    #[error("disk format mismatch: {0}")]
    FormatMismatch(String),
    #[cfg(feature = "fjall")]
    #[error("fjall backend error: {0}")]
    Fjall(String),
    #[cfg(feature = "encryption")]
    #[error("encryption error: {0}")]
    EncryptionError(String),
    #[cfg(feature = "replication")]
    #[error("replication error: {0}")]
    Replication(String),
    #[cfg(feature = "replication")]
    #[error("shard count mismatch: leader has {leader}, follower has {follower}")]
    ShardCountMismatch { leader: usize, follower: usize },
    #[error("catch-up run limit exceeded: observed {actual}, configured {limit}")]
    CatchupResourceLimit { actual: usize, limit: usize },
    /// Schema/version mismatch detected while opening a collection.
    /// See `SchemaMismatchKind` for the specific violation.
    #[error("schema mismatch on '{name}': {kind}")]
    SchemaMismatch {
        name: String,
        kind: SchemaMismatchKind,
    },
    #[error("recovery thread panicked")]
    RecoveryPanic,
}

/// Specific kind of schema mismatch (see `DbError::SchemaMismatch`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SchemaMismatchKind {
    /// Stored and expected `typ_hash` differ at the same `version`.
    /// Indicates the schema changed without bumping `T::VERSION`.
    TypHash { stored: u64, expected: u64 },
    /// Stored `version` is newer than the requested `meta.version`
    /// (rollback to an older binary).
    Downgrade { stored: u16, requested: u16 },
    /// No migration step is registered for the source version `from`.
    MissingStep { from: u16 },
}

impl core::fmt::Display for SchemaMismatchKind {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::TypHash { stored, expected } => write!(
                f,
                "type hash drift: stored={stored:#x} expected={expected:#x}"
            ),
            Self::Downgrade { stored, requested } => write!(
                f,
                "version downgrade: stored={stored} requested={requested}"
            ),
            Self::MissingStep { from } => write!(f, "missing migration from version {from}"),
        }
    }
}

impl DbError {
    pub fn status_code(&self) -> u16 {
        match self {
            DbError::Client(_) => 400,
            DbError::KeyNotFound => 404,
            DbError::NotImplemented => 501,
            DbError::SchemaMismatch { .. } => 422,
            DbError::KeyExists => 409,
            DbError::CasMismatch => 412,
            DbError::TxConflict => 409,
            DbError::TxSchemaMismatch { .. } => 422,
            _ => 500,
        }
    }
}

pub type DbResult<T> = Result<T, DbError>;

impl From<PersistError> for DbError {
    fn from(err: PersistError) -> Self {
        match err {
            PersistError::Io(e) => Self::Io(e),
            PersistError::Json(e) => Self::FormatMismatch(e.to_string()),
        }
    }
}

#[cfg(test)]
mod schema_mismatch_tests {
    use super::*;

    #[test]
    fn schema_mismatch_typ_hash_maps_to_422() {
        let e = DbError::SchemaMismatch {
            name: "users".into(),
            kind: SchemaMismatchKind::TypHash {
                stored: 0xDEAD,
                expected: 0xBEEF,
            },
        };
        assert_eq!(e.status_code(), 422);
    }

    #[test]
    fn schema_mismatch_downgrade_maps_to_422() {
        let e = DbError::SchemaMismatch {
            name: "users".into(),
            kind: SchemaMismatchKind::Downgrade {
                stored: 3,
                requested: 2,
            },
        };
        assert_eq!(e.status_code(), 422);
    }

    #[test]
    fn schema_mismatch_missing_step_maps_to_422() {
        let e = DbError::SchemaMismatch {
            name: "users".into(),
            kind: SchemaMismatchKind::MissingStep { from: 1 },
        };
        assert_eq!(e.status_code(), 422);
    }

    #[test]
    fn key_exists_maps_to_409() {
        assert_eq!(DbError::KeyExists.status_code(), 409);
    }

    #[test]
    fn cas_mismatch_maps_to_412() {
        assert_eq!(DbError::CasMismatch.status_code(), 412);
    }
}

#[cfg(test)]
mod tx_failure_tests {
    use super::*;

    #[test]
    fn tx_failed_after_log_reports_500_and_chains_its_source() {
        let err = DbError::TxFailedAfterLog {
            tx_id: 7,
            stage: TxFailStage::Apply,
            fenced: vec![FencedShard {
                collection: "users",
                shard: 3,
                gsn: 42,
            }],
            source: Box::new(DbError::Internal("injected apply failure")),
        };
        assert_eq!(err.status_code(), 500);

        let text = format!("{err}");
        assert!(text.contains("transaction 7"), "{text}");
        assert!(text.contains("Apply"), "{text}");
        // The wrapper must stay self-contained: existing assertions elsewhere
        // match on the inner error's text through `format!("{err}")`.
        assert!(text.contains("injected apply failure"), "{text}");

        let source = std::error::Error::source(&err).expect("source is chained");
        assert!(format!("{source}").contains("injected apply failure"));
    }

    #[test]
    fn tx_fail_stages_are_distinct() {
        assert_ne!(TxFailStage::Apply, TxFailStage::Finish);
        assert_ne!(TxFailStage::Finish, TxFailStage::Sync);
    }
}