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 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 relation does not resolve on this connection's `search_path` (SQLSTATE `42P01`).
    /// Either `migrate()` has not run, or the connection's `search_path` does not resolve the
    /// unqualified name `outbox` to the migrated schema. **Permanent** — the table does not
    /// appear on its own. Reliar does not check this at construction (ADR 0047); this is the
    /// first statement reporting it, with PostgreSQL's own message attached as the `source`.
    /// Mapped from SQLSTATE `42P01` on **every** call.
    NotMigrated {
        /// The underlying `42P01` error, returned from [`std::error::Error::source`].
        source: sqlx::Error,
    },

    /// 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,
    },
}

impl fmt::Display for PostgresOutboxError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NotMigrated { source } => write!(
                f,
                "the outbox table does not resolve on this connection's search_path: {source}; \
                 run reliar_store_postgres::migrate(&pool, ..) and put the migrated schema first \
                 on search_path — in the connection URL \
                 (options=-c search_path=reliar,public) or with ALTER ROLE <role> SET \
                 search_path = reliar, public"
            ),
            Self::Database { source } => write!(f, "database error: {source}"),
        }
    }
}

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

/// 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::NotMigrated { .. } => FailureKind::Permanent,
            Self::Database { source } => classify_sqlstate(source),
        }
    }
}

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

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

    PostgresOutboxError::Database { source: err }
}

impl crate::error::FromDatabaseError for PostgresOutboxError {
    fn from_database_error(err: sqlx::Error) -> Self {
        map_operational_error(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. A `23505` on `pk_outbox` — which no public path can produce,
/// since `enqueue` never binds `id` and the column defaults to `uuidv7()` — falls through to
/// `Database` in this same `if`, never `Duplicate` (ADR 0049 §2).
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,
    }
}