reliar-store-postgres 0.7.0

//! `complete`/`fail`: the inbox contract's outcome statements. `complete` runs in the caller's
//! transaction; `fail` runs on the provider's own pool, in its own implicit transaction, always
//! guarded by `completed_at IS NULL` so a stale failure can never un-complete a row another
//! consumer finished, and bounds the recorded failure at `settings.max_attempts` atomically with
//! the increment (ADR 0042 A.2.4).

use reliar_core::MessageId;
use reliar_inbox::{InboxFailure, InboxMessage, InboxRecordId, InboxScope};
use sqlx::{Postgres, Transaction};

use super::error::PostgresInboxError;
use crate::connection::schema::{restore_search_path, set_search_path};
use crate::records::truncate_last_error;

use super::PostgresInboxStore;

/// [`reliar_inbox::InboxStore::complete`]'s body. Wraps its `UPDATE` in the same
/// transaction-local `search_path` set/restore [`super::claim::claim`] uses, and for the same
/// reason: with `claim_sets_search_path` on, `claim` restores the caller's `search_path` before
/// returning, so `complete`'s own unqualified `UPDATE inbox` needs its own wrap — leaving it
/// unwrapped would resolve `inbox` against whatever `search_path` the caller's connection had
/// before `claim` ever touched it, which is exactly the case this setting exists to route around.
pub(super) async fn complete(
    store: &PostgresInboxStore,
    tx: &mut Transaction<'_, Postgres>,
    scope: &InboxScope,
    message_id: MessageId,
) -> Result<(), PostgresInboxError> {
    let restore = if store.settings.claim_sets_search_path {
        Some(set_search_path(tx, &store.settings.schema).await?)
    } else {
        None
    };

    let result = complete_locked(tx, scope, message_id).await;

    // Only restore on success, mirroring `claim`'s own wrap: a failed statement already aborts
    // the transaction, so issuing another on it would mask the real error.
    if result.is_ok()
        && let Some(previous) = restore
    {
        restore_search_path(tx, &previous).await?;
    }

    result
}

async fn complete_locked(
    tx: &mut Transaction<'_, Postgres>,
    scope: &InboxScope,
    message_id: MessageId,
) -> Result<(), PostgresInboxError> {
    let scope = scope.as_str();
    let id = message_id.as_uuid();
    // `AND dead_at IS NULL` is load-bearing, not defensive: `ck_inbox_terminal` would otherwise
    // turn "complete a dead row" into a raw check-constraint error instead of a clean
    // `NotClaimed` (ADR 0042 A.2.4).
    let result = sqlx::query!(
        r#"UPDATE inbox SET completed_at = now(), updated_at = now()
            WHERE scope = $1 AND message_id = $2 AND completed_at IS NULL AND dead_at IS NULL"#,
        scope,
        id,
    )
    .execute(&mut **tx)
    .await?;

    if result.rows_affected() == 0 {
        return Err(PostgresInboxError::NotClaimed {
            scope: scope.to_owned(),
            message_id,
        });
    }

    Ok(())
}

/// [`reliar_inbox::InboxStore::fail`]'s body, on `store`'s own pool. `last_error` is the already
/// extracted, already-chained `Display` — see [`format_error_chain`]'s doc for why the caller
/// (the trait impl's `fail`, in block form) must build it *before* this future is constructed:
/// `&(dyn Error + 'static)` is not `Send`, so it cannot cross an `.await` inside the future.
///
/// The `INSERT … ON CONFLICT DO UPDATE` below blocks on a concurrent claimer's still-open
/// `INSERT` of the same `(scope, message_id)` (the redelivery race `claim`'s advisory-lock probe
/// exists to avoid, reintroduced here because `fail` is a plain pool statement, not part of the
/// claiming transaction). [`crate::PostgresInboxSettings::statement_timeout`] bounds that wait
/// exactly as it bounds the outbox's pool-side calls; `fail` is best-effort, so a canceled
/// statement (`FailureKind::Transient`) is fine for the caller to swallow and let the broker
/// redeliver.
pub(super) async fn fail(
    store: &PostgresInboxStore,
    scope: &InboxScope,
    message: InboxMessage<'_>,
    last_error: String,
) -> Result<InboxFailure, PostgresInboxError> {
    let scope_str = scope.as_str();
    let last_error = truncate_last_error(last_error);
    let max_attempts = i32::try_from(store.settings.max_attempts).unwrap_or(i32::MAX);

    let row = if store.settings.statement_timeout.is_zero() {
        fail_row(&store.pool, scope_str, message, &last_error, max_attempts).await?
    } else {
        let mut tx = store.pool.begin().await?;

        store.set_local_timeout(&mut tx).await?;
        let row = fail_row(&mut *tx, scope_str, message, &last_error, max_attempts).await?;
        tx.commit().await?;

        row
    };

    let Some(row) = row else {
        // The completed-row guard fired: zero rows affected.
        return Ok(InboxFailure::AlreadyCompleted);
    };

    let attempts = u32::try_from(row.attempts).unwrap_or(u32::MAX);

    Ok(match row.dead_at {
        Some(dead_at) => InboxFailure::Dead {
            id: InboxRecordId::from_uuid(row.id),
            attempts,
            dead_at,
        },
        None => InboxFailure::Recorded { attempts },
    })
}

/// [`fail`]'s row shape, named via `query_as!` (never `FromRow`).
struct FailedRow {
    id: uuid::Uuid,

    attempts: i32,

    dead_at: Option<time::OffsetDateTime>,
}

/// The `INSERT … ON CONFLICT DO UPDATE … RETURNING` (inbox contract §3.1): the dead transition
/// is computed in SQL (`attempts + 1 >= max_attempts`) so it is atomic with the increment — two
/// concurrent `fail`s cannot race the transition. `$9` binds `max_attempts` for both branches:
/// the insert branch must apply the bound too, or `max_attempts = 1` would never die on its
/// first `fail`. `dead_at`'s `COALESCE(inbox.dead_at, CASE …)` keeps the row's **original**
/// `dead_at` once it is set: without it, a later `fail` on an already-dead row (`attempts + 1 >=
/// max_attempts` still holds) would re-stamp `dead_at = now()` every time, moving the recorded
/// death time forward on every subsequent failed attempt.
async fn fail_row<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    scope: &str,
    message: InboxMessage<'_>,
    last_error: &str,
    max_attempts: i32,
) -> Result<Option<FailedRow>, sqlx::Error> {
    let id = InboxRecordId::new().as_uuid();
    let message_id = message.id.as_uuid();
    let message_type = message.message_type.name();
    let message_version = i32::from(message.message_type.version());
    let conversation_id = message.conversation_id.as_uuid();
    let correlation_id = message
        .correlation_id
        .map(reliar_core::CorrelationId::as_str);
    let causation_id = message.causation_id.map(|c| c.as_uuid());

    sqlx::query_as!(
        FailedRow,
        r#"INSERT INTO inbox (id, scope, message_id, message_type, message_version,
                               conversation_id, correlation_id, causation_id,
                               attempts, last_error, dead_at)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8,
                   1, $9, CASE WHEN $10::integer <= 1 THEN now() END)
           ON CONFLICT (scope, message_id) DO UPDATE
              SET attempts   = inbox.attempts + 1,
                  last_error = excluded.last_error,
                  dead_at    = COALESCE(inbox.dead_at,
                                 CASE WHEN inbox.attempts + 1 >= $10::integer THEN now() END),
                  updated_at = now()
            WHERE inbox.completed_at IS NULL
           RETURNING id, attempts, dead_at"#,
        id,
        scope,
        message_id,
        message_type,
        message_version,
        conversation_id,
        correlation_id,
        causation_id,
        last_error,
        max_attempts,
    )
    .fetch_optional(executor)
    .await
}

/// Joins `error`'s `Display` with every `source()` in its chain, `": "`-separated — the inbox
/// contract's "last failure's error chain". Never touches payload bytes or header values: it
/// only ever sees what the caller's own error type chose to put in its `Display`.
///
/// **Must run before [`fail`]'s future is constructed.** `&(dyn Error + 'static)` is not `Send`
/// (`&T: Send` requires `T: Sync`, and `dyn Error` is not `Sync`), so a plain `async fn fail`
/// holding it across the generator's state — even before any `.await` — fails
/// [`reliar_inbox::InboxStore::fail`]'s `+ Send` bound. This is the synchronous extraction the
/// trait's own rustdoc points implementors to.
pub(super) fn format_error_chain(error: &(dyn std::error::Error + 'static)) -> String {
    let mut out = error.to_string();
    let mut source = error.source();

    while let Some(err) = source {
        out.push_str(": ");
        out.push_str(&err.to_string());
        source = err.source();
    }

    out
}