reliar_store_postgres/

error.rs

//! Hand-rolled error enums for the PostgreSQL provider (ADR 0008).
//!
//! No `thiserror`, no `anyhow`. Every `Display` is payload/credential-free: a decode failure
//! names the message id and a truncated detail, never the offending bytes. Classification is a
//! **per-variant table, never a blanket rule** — a `Database` failure is classified by the
//! wrapped SQLSTATE's class, not assumed transient.

use core::fmt;

use reliar_core::{Classify, FailureKind, MessageId};

/// A failure of a [`crate::PostgresOutboxStore`] `OutboxStore`/`OutboxDeadLetters` *call* —
/// never a property of one row's content. Row-content problems surface as
/// [`reliar_outbox::PoisonedRow`]s instead (ADR 0008).
///
/// [`Classify`] tells a dispatcher whether a failed call is worth retrying:
///
/// ```no_run
/// # async fn run(store: reliar_store_postgres::PostgresOutboxStore) -> Result<(), Box<dyn std::error::Error>> {
/// use reliar_core::Classify;
/// use reliar_outbox::{AcquireRequest, OutboxStore, WorkerId};
///
/// let request = AcquireRequest::new(WorkerId::generate());
/// if let Err(err) = store.acquire(request).await {
///     eprintln!("acquire failed ({:?}): {err}", err.kind());
/// }
/// # Ok(())
/// # }
/// ```
#[derive(Debug)]
#[non_exhaustive]
pub enum PostgresStoreError {
    /// The unqualified name `outbox` does not resolve, or resolves to a different schema than
    /// configured. Carries the configured schema and the observed `search_path`; the `ALTER
    /// ROLE` remedy is in the `Display` text. **Permanent.**
    SchemaResolution {
        /// The schema `PostgresOutboxSettings::schema` named.
        configured: String,
        /// The `search_path` Postgres reported at construction.
        observed: String,
    },

    /// `outbox` resolved to the configured schema, but the relation itself is missing —
    /// `migrate()` has not been run. **Permanent.** Mapped from SQLSTATE `42P01` on **every**
    /// path, not just startup verification.
    NotMigrated {
        /// The configured schema.
        schema: String,
    },

    /// Connection lost, statement timeout, pool exhausted, deadlock, or any other `sqlx`
    /// failure not mapped to a more specific variant above. Classified by the wrapped
    /// SQLSTATE's **class** (never blanket-transient — see the `Classify` impl below).
    Database {
        /// The underlying `sqlx` error.
        source: sqlx::Error,
    },

    /// A claimed or listed row could not be turned into an `OutboxRecord` (a corrupt JSONB
    /// remainder, an unparseable promoted column). Surfaces as a poisoned row, never as an
    /// `acquire`/`list_dead` failure. **Permanent** — the bytes on disk do not change between
    /// attempts.
    Decode {
        /// The row's message id.
        id: MessageId,
        /// A short, payload-free description of what failed to decode.
        detail: String,
    },

    /// The row's `metadata_version` is not one this build knows how to read. **Permanent** —
    /// it needs a newer reader, not another try.
    UnknownMetadataVersion {
        /// The row's message id.
        id: MessageId,
        /// The unrecognised version.
        version: i32,
    },

    /// `enqueue` inserted a `MessageId` that already exists (`pk_outbox` violation).
    /// **Permanent** — a reused id never succeeds on retry; the row is already there.
    DuplicateMessage {
        /// The id the caller tried to reuse.
        id: MessageId,
    },

    /// `PostgresOutboxSettings::schema` or `MigrateOptions::schema` is not a valid PostgreSQL
    /// identifier (`[a-z_][a-z0-9_$]*`, at most 63 bytes, **lowercase only**) — checked once,
    /// before it is ever interpolated into `SET search_path`/`dangerous_set_table_name`.
    /// Lowercase-only rather than merely case-insensitive: PostgreSQL folds an
    /// *unquoted* identifier to lowercase, so an uppercase configured name and the schema it
    /// actually resolves to would silently disagree unless every one of `migrate()`'s,
    /// `verify_schema`'s and the host's own `search_path` configuration happened to quote it the
    /// same way everywhere — rejecting it up front removes the whole class of mismatch.
    /// **Permanent** — configuration, not weather.
    InvalidSchema {
        /// The rejected schema name.
        schema: String,
    },

    /// The connected server's `server_version_num` is below [`crate::MIN_SERVER_VERSION_NUM`]
    /// (PostgreSQL 18, ADR 0041 / human decision #47) — **no older-version fallback**. Checked
    /// at [`crate::PostgresOutboxStore::connect`], **before** the `search_path` verification
    /// above: a wrong server version explains a missing relation, and the reverse is never
    /// true. Carries no connection string, host, or credentials. **Permanent.**
    UnsupportedServerVersion {
        /// [`crate::MIN_SERVER_VERSION_NUM`], restated on the value so this variant is
        /// self-describing without a second lookup.
        required: u32,
        /// The `server_version_num` this connection reported.
        detected: u32,
    },
}

impl fmt::Display for PostgresStoreError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::SchemaResolution {
                configured,
                observed,
            } => write!(
                f,
                "outbox did not resolve to schema \"{configured}\" (observed search_path: \
                 \"{observed}\"); set search_path so \"{configured}\" comes first, e.g. \
                 ALTER ROLE <role> SET search_path = {configured}, public"
            ),
            Self::NotMigrated { schema } => write!(
                f,
                "relation \"{schema}.outbox\" does not exist; call \
                 reliar_store_postgres::migrate(&pool, ..) before constructing the store"
            ),
            Self::Database { source } => write!(f, "database error: {source}"),
            Self::Decode { id, detail } => write!(f, "row {id} could not be decoded: {detail}"),
            Self::UnknownMetadataVersion { id, version } => {
                write!(f, "row {id} carries unknown metadata_version {version}")
            }
            Self::DuplicateMessage { id } => {
                write!(f, "message id {id} already exists in the outbox")
            }
            Self::InvalidSchema { schema } => write!(
                f,
                "{schema:?} is not a valid PostgreSQL identifier (expected \
                 [a-z_][a-z0-9_$]*, at most 63 bytes, lowercase only — PostgreSQL folds an \
                 unquoted identifier to lowercase, so an uppercase name would resolve \
                 inconsistently)"
            ),
            Self::UnsupportedServerVersion { required, detected } => write!(
                f,
                "PostgreSQL 18 or newer is required (server_version_num >= {required}); \
                 detected {detected} — there is no supported way to run Reliar below the floor"
            ),
        }
    }
}

impl std::error::Error for PostgresStoreError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Database { source } => Some(source),
            _ => None,
        }
    }
}

/// Per-variant classification table — **no blanket "everything else is
/// transient"**. A wrong verdict is not cosmetic: `Transient` burns the dispatcher's retry
/// budget on a failure that can never succeed; `Permanent` kills a message that would have gone
/// through on the next attempt.
impl Classify for PostgresStoreError {
    fn kind(&self) -> FailureKind {
        match self {
            Self::SchemaResolution { .. }
            | Self::NotMigrated { .. }
            | Self::InvalidSchema { .. }
            | Self::DuplicateMessage { .. }
            | Self::Decode { .. }
            | Self::UnknownMetadataVersion { .. }
            | Self::UnsupportedServerVersion { .. } => FailureKind::Permanent,
            Self::Database { source } => classify_sqlstate(source),
        }
    }
}

/// Classifies a `sqlx::Error` by its wrapped SQLSTATE **class**, never by message text:
///
/// - **Transient** — `08*` (connection exception), `40*` (transaction rollback: deadlock,
///   serialization failure), `53*` (insufficient resources), `55*` (object in use), `57014`
///   (`query_canceled`, i.e. a `statement_timeout`), and any pool/IO error with no SQLSTATE at all.
/// - **Permanent** — `22*` (data exception), `23*` (integrity constraint violation), `42*`
///   (syntax error or access rule violation — includes `42P01`, mapped to `NotMigrated` before
///   this function ever sees it).
/// - Anything unrecognised classifies **Transient** — an unknown fault is more often weather
///   than logic — but is logged at `warn` with its SQLSTATE so this table can be extended.
pub(crate) fn classify_sqlstate(err: &sqlx::Error) -> FailureKind {
    let sqlx::Error::Database(db) = err else {
        // No SQLSTATE at all: a connection/IO/pool-exhaustion failure, not a data problem.
        return FailureKind::Transient;
    };
    let Some(code) = db.code() else {
        return FailureKind::Transient;
    };

    match code.as_ref().get(..2) {
        Some("08" | "40" | "53" | "55") => FailureKind::Transient,
        Some("57") if code.as_ref() == "57014" => FailureKind::Transient,
        Some("22" | "23" | "42") => FailureKind::Permanent,
        _ => {
            tracing::warn!(sqlstate = %code, "unrecognised SQLSTATE; classifying transient");

            FailureKind::Transient
        }
    }
}

impl From<sqlx::Error> for PostgresStoreError {
    fn from(source: sqlx::Error) -> Self {
        Self::Database { source }
    }
}

/// Maps a `sqlx::Error` to a typed error, keying on SQLSTATE and constraint **name** — never on
/// message text — so `42P01` maps to `NotMigrated` **on every path**, not just startup
/// verification, and everything else falls through to `Database` for
/// [`classify_sqlstate`] to classify.
pub(crate) fn map_operational_error(schema: &str, err: sqlx::Error) -> PostgresStoreError {
    if is_undefined_table(&err) {
        return PostgresStoreError::NotMigrated {
            schema: schema.to_owned(),
        };
    }

    PostgresStoreError::Database { source: err }
}

/// [`crate::PostgresOutboxStore`]'s [`reliar_outbox::OutboxEnqueue::enqueue_envelope`] failures. Enqueuing
/// runs on the **host's** write path, where the host decides whether to retry its own
/// transaction, so this implements [`Classify`] on the same rules as [`PostgresStoreError`]
/// rather than making the host re-derive which SQLSTATEs are worth retrying.
///
/// A duplicate [`reliar_core::MessageId`] aborts the caller's transaction rather than silently
/// losing the message:
///
/// ```no_run
/// # async fn run(
/// #     store: reliar_store_postgres::PostgresOutboxStore,
/// #     pool: sqlx::PgPool,
/// # ) -> Result<(), Box<dyn std::error::Error>> {
/// use reliar_core::{Classify, Message};
/// use reliar_outbox::OutboxEnqueue;
///
/// #[derive(serde::Serialize, serde::Deserialize)]
/// struct OrderPlaced;
/// impl Message for OrderPlaced {
///     const TYPE: &'static str = "orders.placed";
///     const VERSION: u16 = 1;
/// }
///
/// let mut tx = pool.begin().await?;
/// if let Err(err) = store.enqueue(&mut tx, OrderPlaced).await {
///     eprintln!("enqueue failed ({:?}): {err}", err.kind());
///     tx.rollback().await?;
/// }
/// # Ok(())
/// # }
/// ```
#[derive(Debug)]
#[non_exhaustive]
pub enum EnqueueError<E> {
    /// The configured [`reliar_core::Serializer`] rejected the body. **Permanent** — the same
    /// body serializes the same way every time.
    Serialize {
        /// The serializer's own error.
        source: E,
    },

    /// The envelope's `MessageId` already exists (`pk_outbox` violation) — `enqueue` uses a
    /// plain `INSERT` with no `ON CONFLICT`, so a reused id aborts the caller's transaction
    /// rather than silently losing a message. **Permanent** — the id is already taken.
    Duplicate {
        /// The id the caller tried to reuse.
        id: MessageId,
    },

    /// Any other `sqlx` failure, classified by SQLSTATE exactly as
    /// [`PostgresStoreError::Database`] (including `42P01`, which classifies permanent under
    /// the `42*` rule).
    Database {
        /// The underlying `sqlx` error.
        source: sqlx::Error,
    },
}

impl<E: fmt::Display> fmt::Display for EnqueueError<E> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Serialize { source } => {
                write!(f, "failed to serialize the envelope body: {source}")
            }
            Self::Duplicate { id } => write!(f, "message id {id} already exists in the outbox"),
            Self::Database { source } => write!(f, "database error: {source}"),
        }
    }
}

impl<E: std::error::Error + 'static> std::error::Error for EnqueueError<E> {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Serialize { source } => Some(source),
            Self::Database { source } => Some(source),
            Self::Duplicate { .. } => None,
        }
    }
}

impl<E: std::error::Error + Send + Sync + 'static> Classify for EnqueueError<E> {
    fn kind(&self) -> FailureKind {
        match self {
            Self::Serialize { .. } | Self::Duplicate { .. } => FailureKind::Permanent,
            Self::Database { source } => classify_sqlstate(source),
        }
    }
}

/// Maps a `sqlx::Error` from an `enqueue` `INSERT` to a typed error, keying on the constraint
/// **name** — never on message text — so `pk_outbox` maps to `Duplicate` and every
/// other failure (including `42P01`) stays `Database`, for [`classify_sqlstate`] to classify.
pub(crate) fn map_enqueue_error<E>(id: MessageId, err: sqlx::Error) -> EnqueueError<E> {
    if is_constraint_violation(&err, "pk_outbox") {
        return EnqueueError::Duplicate { id };
    }

    EnqueueError::Database { source: err }
}

/// `true` when `err` is a unique/check-constraint violation on `constraint`. Keys on the
/// **name**, never on message text — that naming discipline is what keeps this map stable
/// across PostgreSQL versions.
pub(crate) fn is_constraint_violation(err: &sqlx::Error, constraint: &str) -> bool {
    match err {
        sqlx::Error::Database(db) => db.constraint() == Some(constraint),
        _ => false,
    }
}

/// `true` for SQLSTATE `42P01` (`undefined_table`) — the relation is missing, i.e. `migrate()`
/// has not run.
pub(crate) fn is_undefined_table(err: &sqlx::Error) -> bool {
    match err {
        sqlx::Error::Database(db) => db.code().as_deref() == Some("42P01"),
        _ => false,
    }
}

/// Validates a schema name against PostgreSQL's unquoted-identifier grammar, restricted to
/// **lowercase** (`[a-z_][a-z0-9_$]*`, at most 63 bytes — Postgres's own `NAMEDATALEN` limit)
/// **before** it is ever interpolated into `SET search_path`/`dangerous_set_table_name`, both of
/// which build SQL text from this value rather than binding it as data. Used by
/// both `PostgresOutboxSettings::schema` (at `connect`) and `MigrateOptions::schema` (at
/// `migrate`), so the two validate identically.
///
/// **Lowercase only, not merely case-insensitive (ADR 0040 §5).** PostgreSQL folds an *unquoted*
/// identifier to lowercase, so `schema = "Foo"` would migrate into a schema literally named
/// `"Foo"` (quoted) while every unqualified reference — the claim, `stats()`, the host's own
/// `search_path` — resolves the unquoted, lowercase-folded `foo` instead: a mismatch this crate
/// cannot detect from inside a single connection's `search_path`, since the host's own connection
/// string or `ALTER ROLE` also has to agree, and cannot be fixed here. Rejecting every uppercase
/// character removes the class of mismatch instead of chasing it through four call sites.
pub(crate) fn is_valid_schema_name(schema: &str) -> bool {
    let mut chars = schema.chars();
    let Some(first) = chars.next() else {
        return false;
    };

    if !(first.is_ascii_lowercase() || first == '_') {
        return false;
    }

    schema.len() <= 63
        && chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '$')
}