armdb 0.11.2

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("guarded compaction requires hints for shard {shard}")]
    GuardedCompactionRequiresHints { shard: u8 },
    #[error("CAS value mismatch")]
    CasMismatch,
    #[error("key already exists")]
    KeyExists,
    /// A queue acknowledgement or retry used a stale lease token.
    #[error("queue lease expired or was superseded")]
    LeaseExpired,
    /// A queue's configured soft depth limit refused a push.
    #[error("queue is full")]
    QueueFull,
    /// A source with a change-stream declaration was mutated outside `db.tx()`.
    #[error("change stream from `{collection}` to queue `{queue}` requires db.tx()")]
    ChangeStreamRequiresTx {
        collection: &'static str,
        queue: &'static str,
    },
    /// A stream-carrying collection was opened without a managed database.
    #[error("change-stream carriers require a managed Db::open_*, not a direct open")]
    ChangeStreamRequiresManagedOpen,
    /// A watched Map was opened without `Config::iterable(true)`. Refused at
    /// open so that `watch()` itself can stay infallible: the recipe a receiver
    /// depends on — subscribe, scan, rescan on `Lagged` — needs a scan, and a
    /// Map has none without an iterator view.
    #[error("a watched Map requires Config::iterable(true)")]
    WatchRequiresIterable,
    /// A watch-carrying hook was attached to an `LsmTree`, which is not a
    /// mutation-seam participant and would silently never emit.
    #[error("realtime watches are unsupported on LsmTree")]
    WatchUnsupported,
    /// 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,
    },
    #[error(
        "pending transaction format is ambiguous on '{collection}' (tx_id={tx_id}, fingerprint={fingerprint:#x}); the log does not qualify codec versions"
    )]
    TxFormatAmbiguous {
        collection: &'static str,
        tx_id: u64,
        fingerprint: 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 },
    /// A claim was emitted for a unique namespace that is not a declared
    /// target of this source's [`crate::UniqueTargets`] (T10).
    #[error("undeclared unique target: {namespace}")]
    UndeclaredUniqueTarget { namespace: &'static str },
    /// A non-transactional write (facade or `atomic()`) would change
    /// `collection`'s claim set in unique namespace `namespace` (T10 Task 5).
    /// Refused **before** any mutation — the write was not applied. Uniqueness
    /// is enforced only inside `db.tx()`; retry there.
    #[error(
        "collection `{collection}` attempted a non-transactional write that changes its claim \
         set in unique namespace `{namespace}` — retry inside db.tx()"
    )]
    UniqueIndexRequiresTx {
        collection: &'static str,
        namespace: &'static str,
    },
    /// Inside `db.tx()`, a claim resolved (`unique::resolve::decide`, T10
    /// Task 7) to "someone else holds this key": the whole transaction is
    /// refused, before the log append — nothing it wrote lands, including
    /// the source row that emitted the claim. `key` and `owner` are rendered
    /// as strings (`key` hex via `crate::hex::hex`, the same renderer the
    /// counter errors use) rather than typed, since this layer only knows a
    /// namespace row's raw key bytes and the claimant's opaque collection
    /// tag, not either side's application type.
    ///
    /// A `Release` conflicts under this same rule when the row it targets is
    /// held by a *different* source row than the one releasing it — the
    /// resolver never deletes a row it does not own (`unique::resolve::
    /// decide`'s foreign-owner arm covers every `ClaimKind`, not just
    /// `Claim`). This is reachable on data that predates T10's enforcement:
    /// if two rows both already claimed the same key before the namespace
    /// existed, the one that does not "win" the namespace row can never
    /// again release *or* re-claim it inside `db.tx()` — every attempt
    /// refuses with this same error, permanently, since nothing here
    /// mutates the offending source row. That is a known, expected
    /// consequence of adopting a unique namespace over pre-existing data,
    /// not a bug; recovering from it is the schema validator/repair path's
    /// job (surface the dangling claim, then fix the source data directly),
    /// not something a transaction can resolve on its own.
    #[error("unique namespace `{namespace}`: key {key} is already held by {owner}")]
    UniqueViolation {
        namespace: &'static str,
        key: String,
        owner: String,
    },
    /// A reservation's TTL failed validation (T10 Task 8). `reason` is one of:
    ///
    /// * `"zero duration"` — [`crate::txn::Tx::reserve`] rejects
    ///   `Duration::ZERO` rather than write an already-expired row: that state
    ///   is indistinguishable from a leak, and the caller means
    ///   [`crate::txn::Tx::release`] instead.
    /// * `"deadline overflow"` — either the requested TTL's millisecond count
    ///   does not fit a `u64` (checked in the builder, before any lock), or
    ///   `now + ttl` does not (checked in the resolver, under the
    ///   namespace's own lock, at `commit()`).
    #[error("unique namespace `{namespace}`: invalid reservation — {reason}")]
    UniqueReservationInvalid {
        namespace: &'static str,
        reason: &'static str,
    },
    /// A contribution was emitted for a collection that is not a declared
    /// counter target of this source.
    #[error("undeclared counter target `{target}`")]
    UndeclaredCounterTarget { target: &'static str },
    /// No target of a `one_of` group holds the row the contribution names.
    #[error("counter target row missing: key {key} in one of [{targets}]")]
    CounterTargetMissing { targets: String, key: String },
    /// More than one target of a `one_of` group holds the row.
    #[error("ambiguous counter target: key {key} exists in [{targets}]")]
    AmbiguousCounterTarget { targets: String, key: String },
    /// Checked counter arithmetic left the `u64` range (either direction).
    #[error("counter `{collection}`.`{field}` overflow at key {key}")]
    CounterOverflow {
        collection: &'static str,
        field: &'static str,
        key: String,
    },
    /// A carrier with counter targets was opened without a `Db`.
    #[error("counter carriers require a managed `Db::open_*`, not a direct open")]
    CountersRequireManagedOpen,
    /// A carrier with claim targets was opened without a `Db` (T10 review I6).
    /// A direct open is never `bind_tx_arc`'d, so it can never join `db.tx()`
    /// — every claim-changing write on such a collection would be refused
    /// forever by `UniqueIndexRequiresTx`. Mirrors [`Self::CountersRequireManagedOpen`].
    #[error("claim carriers require a managed `Db::open_*`, not a direct open")]
    ClaimsRequireManagedOpen,
    /// A carrier with claim targets was opened on a collection whose value
    /// type declares a TTL (T10 review C1). `SchemaRegistry::unique_index_on`
    /// already asserts this at declaration time, but `Projected::and_unique`,
    /// `UniqueTargets` and `Db::open_unique_index` are all `pub`, so a
    /// hand-assembled carrier can reach a managed open without ever meeting
    /// that assert. Left unrejected, the sweeper's guard placement disagrees
    /// by family: `TypedTree`/`TypedMap`/`ZeroTree`/`ZeroMap` retry the
    /// expiry delete forever (`UniqueIndexRequiresTx` fires *inside* the
    /// locked delete), while `VarTypedTree`/`VarTypedMap` delete the row and
    /// silently orphan its claim — a name locked forever with no owner,
    /// exactly what decision 13 in the spec exists to prevent.
    #[error(
        "unique-namespace claiming source `{collection}` declares a TTL — a sweep runs outside \
         any transaction and could not release a claim atomically"
    )]
    ClaimingSourceHasTtl { collection: &'static str },
    /// A carrier with claim targets was opened on a collection whose own key
    /// does not fit a unique namespace owner row (T10 review F3).
    /// `SchemaRegistry::unique_index_on` already asserts
    /// `size_of::<SV::SelfId>() <= UNIQUE_OWNER_KEY_CAP` at declaration time,
    /// but — mirroring [`Self::ClaimingSourceHasTtl`] — that assert is only
    /// reachable through the registry path: `Projected::and_unique`,
    /// `UniqueTargets` and every managed `Db::open_*` are all `pub`, so a
    /// hand-assembled carrier can reach a managed open without ever meeting
    /// it. Left unrejected, the collection opens fine and then fails every
    /// claim-changing transaction with `DbError::Internal("claim source key
    /// exceeds the owner capacity")` from `into_tx_claim_ops`.
    #[error(
        "unique-namespace claiming source `{collection}`'s key does not fit a unique namespace \
         owner row (56 bytes)"
    )]
    ClaimingSourceKeyTooWide { collection: &'static str },
    /// Two distinct source collections declared claims into the same unique
    /// namespace and their names hash to the same 64-bit FNV owner tag (T10
    /// review F5). `SchemaRegistry::unique_index`/`unique_index_on` already
    /// runs this exact check (`register_owner_tag`) at declaration time, but
    /// it has no way to see a source that reaches a managed `Db::open_*`
    /// through the raw `Projected::and_unique` path with no `SchemaRegistry`
    /// in the picture at all. Left unrejected, `UniqueOwner::same_holder`
    /// would read the two sources' rows as the same owner: the second source
    /// could claim an already-owned key with no violation, and either could
    /// release the other's claim. Accidental collision is ~2^-64 — this is
    /// defence-in-depth, not a live bug — but the check already exists and is
    /// cheap to run again at the one other place a claiming source is opened.
    #[error(
        "unique namespace `{namespace}` owner tag collision — sources `{a}` and `{b}` hash to \
         the same collection tag"
    )]
    ClaimOwnerTagCollision {
        namespace: &'static str,
        a: &'static str,
        b: &'static str,
    },
    /// [`crate::armour::Db::recount_counters`] was called on a collection
    /// that declares no counter targets of its own (it may be a plain
    /// counter *target*, or unrelated to the counter system entirely).
    /// Recounting it would be a silent no-op that still clears its
    /// `counters_dirty` flag — for a target, that flag is the only evidence
    /// a source's own later open reads to know a recount is owed, so this is
    /// refused rather than done quietly.
    #[error("`{collection}` declares no counter targets — not a counted source")]
    NotACounterSource { collection: String },
    /// 400 — a transaction wrote directly to a collection that is *both* a
    /// declared secondary-index/counter **target** and a projecting or
    /// counting **source** in its own right.
    ///
    /// A target is locked in the commit's **second** round, and that round
    /// evaluates writes without projecting them
    /// (`txn::projection_commit::eval_rank1` collects no projection inputs).
    /// So such a write would land, and the collection's *own* indexes and
    /// counters would silently stop being maintained — drift with no error and
    /// no log line, discoverable only by a later `schema::validate` run.
    /// Refused up front instead.
    #[error(
        "`{collection}` is a projection/counter target that is also a projecting source of its \
         own, so it cannot be written directly inside a transaction: a target is locked in the \
         commit's second round, which does not run projections, and its own indexes/counters \
         would silently stop being maintained. Write it through the collection API (outside \
         `db.tx()`), or move its projections onto a collection that is not itself a target."
    )]
    TxTargetIsAlsoASource { collection: &'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("format migration required for `{collection}` from {from} to {to}")]
    FormatMigrationRequired {
        collection: &'static str,
        from: u16,
        to: u16,
    },
    #[error("duplicate migration step for `{collection}` at version {version}")]
    DuplicateMigrationStep {
        collection: &'static str,
        version: u16,
    },
    #[error("format migration for `{collection}` at version {version} follows a value migration")]
    FormatStepAfterValueStep {
        collection: &'static str,
        version: u16,
    },
    #[error(
        "format type mismatch for `{collection}` at version {from}: expected {expected:#x}, stored {stored:#x}"
    )]
    FormatTypMismatch {
        collection: &'static str,
        from: u16,
        expected: u64,
        stored: u64,
    },
    #[error(
        "format migration step for `{collection}` from version {from_version} failed at key {key}: {cause}"
    )]
    FormatStepFailed {
        collection: &'static str,
        from_version: u16,
        key: String,
        cause: String,
    },
    #[error("format migration for `{collection}` shard {shard} needs {needed} fresh file ids")]
    FormatFileIdExhausted {
        collection: &'static str,
        shard: usize,
        needed: usize,
    },
    #[error("format migration is unsupported by backend `{backend}` for `{collection}`")]
    FormatMigrationUnsupported {
        collection: &'static str,
        backend: &'static str,
    },
    #[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::NotACounterSource { .. } => 400,
            DbError::TxTargetIsAlsoASource { .. } => 400,
            DbError::UniqueReservationInvalid { .. } => 400,
            DbError::KeyNotFound => 404,
            DbError::NotImplemented => 501,
            DbError::SchemaMismatch { .. } => 422,
            DbError::KeyExists => 409,
            DbError::UniqueViolation { .. } => 409,
            DbError::CasMismatch => 412,
            DbError::TxConflict => 409,
            DbError::TxSchemaMismatch { .. } | DbError::TxFormatAmbiguous { .. } => 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);
    }
}