reliar_store_postgres/outbox/

error.rs

//! Hand-rolled error enums for the outbox side of 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};
use reliar_outbox::OutboxRecordId;

use crate::error::{classify_sqlstate, is_undefined_table};

/// 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. The bare
/// `PostgresOutboxStore` below leans on its default type parameter, gated on the default
/// `json` feature; without it this block still shows the shape but is not compiled.
#[cfg_attr(not(feature = "json"), doc = "```ignore")]
#[cfg_attr(feature = "json", doc = "```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 PostgresOutboxError {
    /// 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.**
    SchemaNotOnSearchPath {
        /// 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,
    },

    /// `outbox` resolved to the configured schema and the relation exists, but it has not
    /// finished a required migration — `missing` names the first column that is either absent or
    /// present but still nullable (a **completion marker**, not a bare inventory check: `id`
    /// exists from migration `0005` onward but stays nullable until `0010`'s `SET NOT NULL`, so a
    /// schema stuck anywhere in `0005`–`0009` is reported the same as one stuck at `0004`).
    /// Checked at [`crate::PostgresOutboxStore::connect`], **after** the `search_path`
    /// verification above and only once the relation is confirmed to exist there: a wrong
    /// `search_path` or a missing relation each already has its own variant, so this one means
    /// specifically "the right table, an old shape" (ADR 0044 Amendment A.4, marker corrected by
    /// Amendment A.5) — today, `message_id` or `id` (migrations `0005`–`0010`). The remedy is
    /// `migrate(&pool, ..)`, never a `search_path` fix. **Permanent** — the column will not
    /// satisfy itself.
    SchemaOutOfDate {
        /// The configured schema.
        schema: String,
        /// The first required column this build did not find satisfied (absent, or present but
        /// still nullable) on the resolved relation.
        missing: &'static str,
    },

    /// 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. Carries both ids (ADR 0044 A.2) — `id` and `message_id` are plain `uuid` columns
    /// and are always readable even when the envelope columns that failed to decode are not.
    Decode {
        /// The row's own identity.
        id: OutboxRecordId,
        /// The row's message id.
        message_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. Carries both ids, see [`Self::Decode`].
    UnknownMetadataVersion {
        /// The row's own identity.
        id: OutboxRecordId,
        /// The row's message id.
        message_id: MessageId,
        /// The unrecognised version.
        version: i32,
    },

    /// `enqueue` inserted a `MessageId` that already exists (`ix_outbox_message_id` violation,
    /// ADR 0044 §1). **Permanent** — a reused id never succeeds on retry; the row is already
    /// there. Unlike [`Self::Decode`]/[`Self::UnknownMetadataVersion`] this carries only the
    /// message id: the caller already knows it, and the row it collided with is not this call's
    /// concern (ADR 0044 A.2 — a `pk_outbox` collision, a *record*-id repeat, is a different,
    /// non-caller error and stays `Database`).
    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,
    /// this crate's own schema check'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) — **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 PostgresOutboxError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::SchemaNotOnSearchPath {
                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::SchemaOutOfDate { schema, missing } => write!(
                f,
                "relation \"{schema}.outbox\" does not satisfy required column \"{missing}\" \
                 (absent, or present but still nullable — migrations 0005-0010 have not all \
                 completed); run reliar_store_postgres::migrate(&pool, ..) before constructing \
                 the store"
            ),
            Self::Database { source } => write!(f, "database error: {source}"),
            Self::Decode {
                id,
                message_id,
                detail,
            } => write!(
                f,
                "row {id} (message {message_id}) could not be decoded: {detail}"
            ),
            Self::UnknownMetadataVersion {
                id,
                message_id,
                version,
            } => write!(
                f,
                "row {id} (message {message_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 PostgresOutboxError {
    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 PostgresOutboxError {
    fn kind(&self) -> FailureKind {
        match self {
            Self::SchemaNotOnSearchPath { .. }
            | Self::NotMigrated { .. }
            | Self::SchemaOutOfDate { .. }
            | Self::InvalidSchema { .. }
            | Self::DuplicateMessage { .. }
            | Self::Decode { .. }
            | Self::UnknownMetadataVersion { .. }
            | Self::UnsupportedServerVersion { .. } => FailureKind::Permanent,
            Self::Database { source } => classify_sqlstate(source),
        }
    }
}

impl From<sqlx::Error> for PostgresOutboxError {
    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) -> PostgresOutboxError {
    if is_undefined_table(&err) {
        return PostgresOutboxError::NotMigrated {
            schema: schema.to_owned(),
        };
    }

    PostgresOutboxError::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 [`PostgresOutboxError`]
/// 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. The bare `PostgresOutboxStore` below leans on its default type
/// parameter, gated on the default `json` feature; without it this block still shows the shape
/// but is not compiled.
#[cfg_attr(not(feature = "json"), doc = "```ignore")]
#[cfg_attr(feature = "json", doc = "```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 (`ix_outbox_message_id` violation, ADR 0044
    /// §1) — `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
    /// [`PostgresOutboxError::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 `ix_outbox_message_id` maps to `Duplicate` (ADR 0044
/// §1: `enqueue` never binds `id`, so the only conflict an `INSERT` can hit is a reused
/// `message_id`) 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, "ix_outbox_message_id") {
        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,
    }
}