reliar_store_postgres/inbox/

inbox_store.rs

//! The [`PostgresInboxStore`] type itself: fields, construction (`connect`), and the
//! [`reliar_inbox::InboxStore`] implementation, which delegates each method's body to its concern
//! module (`claim`, `outcomes`, `purge`) — mirroring `outbox::outbox_store`'s shape.

use reliar_core::MessageId;
use reliar_inbox::InboxStore;
use reliar_inbox::{
    InboxClaim, InboxFailure, InboxMessage, InboxPurgeReport, InboxPurgeRequest, InboxRecord,
    InboxScope,
};
use sqlx::{PgPool, Postgres, Transaction};
use tracing::Instrument as _;

use crate::connection::schema;
use crate::settings::PostgresInboxSettings;

use super::error::PostgresInboxError;
use super::{claim, outcomes, purge};

/// Reliar's PostgreSQL inbox provider (inbox contract §3). A **separate type** from
/// [`crate::PostgresOutboxStore`]: the inbox stores no payload, so it needs no `Serializer` type
/// parameter and none of the outbox's lease/ordering/retention settings. Same crate, same schema,
/// same [`crate::migrate`]. Cheap to clone — wraps a [`PgPool`]; no outer `Arc` required.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct PostgresInboxStore {
    pub(super) pool: PgPool,

    pub(super) settings: PostgresInboxSettings,
}

impl PostgresInboxStore {
    /// Verifies the server version (ADR 0041) and then the `search_path` (ADR 0018/§20.1), in
    /// that order — exactly as [`crate::PostgresOutboxStore::connect`] does: a wrong server
    /// version explains a missing relation, and the reverse is never true. Logs a
    /// `tracing::warn!` when a same-named table also exists in another schema on the path.
    ///
    /// # Errors
    ///
    /// Returns [`PostgresInboxError::UnsupportedServerVersion`],
    /// [`PostgresInboxError::NotMigrated`] (the relation itself is missing),
    /// [`PostgresInboxError::SchemaNotOnSearchPath`] (it exists, but not on the configured
    /// schema), or [`PostgresInboxError::Database`] for a connection failure during verification.
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::{PostgresInboxSettings, PostgresInboxStore};
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresInboxStore::connect(pool, PostgresInboxSettings::default()).await?;
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect(
        pool: PgPool,
        settings: PostgresInboxSettings,
    ) -> Result<Self, PostgresInboxError> {
        settings.validate()?;

        let detected = crate::connection::version::detected_server_version_num(&pool).await?;

        if detected < crate::MIN_SERVER_VERSION_NUM {
            return Err(PostgresInboxError::UnsupportedServerVersion {
                required: crate::MIN_SERVER_VERSION_NUM,
                detected,
            });
        }

        let check = schema::verify_table_schema(&pool, &settings.schema, "inbox", &[]).await?;
        let resolved_here = check.resolved_schema.as_deref() == Some(settings.schema.as_str());

        if !resolved_here {
            // Mirrors `PostgresOutboxStore::connect`'s own ordering: a missing relation is a
            // sharper, more actionable answer than "wrong search_path" when it is in fact the
            // cause — `migrate()` never ran, rather than merely resolving somewhere else.
            if !check.configured_exists {
                return Err(PostgresInboxError::NotMigrated {
                    schema: settings.schema,
                });
            }

            return Err(PostgresInboxError::SchemaNotOnSearchPath {
                configured: settings.schema,
                observed: check.search_path,
            });
        }

        let others = schema::other_table_schemas(&pool, &settings.schema, "inbox").await?;

        if !others.is_empty() {
            tracing::warn!(
                configured_schema = %settings.schema,
                other_schemas = ?others,
                "a table named `inbox` also exists outside the configured schema; an \
                 unqualified reference from another session could resolve to it"
            );
        }

        Ok(Self { pool, settings })
    }

    /// Issues `SET LOCAL statement_timeout` on an already-open transaction — the shared half of
    /// every `Duration::ZERO`-vs-non-zero split in `outcomes::fail` and `purge`'s pool-side
    /// statements, mirroring [`crate::PostgresOutboxStore::set_local_timeout`].
    pub(super) async fn set_local_timeout(
        &self,
        tx: &mut Transaction<'_, Postgres>,
    ) -> Result<(), PostgresInboxError> {
        let timeout_ms = i64::try_from(self.settings.statement_timeout.as_millis())
            .unwrap_or(i64::MAX)
            .to_string();

        sqlx::query_scalar!(
            "SELECT set_config('statement_timeout', $1, true)",
            timeout_ms
        )
        .fetch_one(&mut **tx)
        .await?;

        Ok(())
    }
}

impl<'c> InboxStore<Transaction<'c, Postgres>> for PostgresInboxStore {
    type Error = PostgresInboxError;

    /// The three-statement claim (inbox contract §3.1): the in-flight advisory-lock guard, the
    /// `INSERT … ON CONFLICT DO NOTHING` claim, and — only when nothing was inserted — the state
    /// read that decides `AlreadyCompleted`/`Dead` vs. `Claimed`.
    // Block form — reason (a): `inbox.scope`/`message.id`/`message.type` must be recorded on the
    // `reliar.inbox.claim` span before the first statement runs (inbox contract §4), so the span
    // itself has to exist before the async block that runs those statements does.
    fn claim(
        &self,
        tx: &mut Transaction<'c, Postgres>,
        scope: &InboxScope,
        message: InboxMessage<'_>,
    ) -> impl Future<Output = Result<InboxClaim, Self::Error>> + Send {
        let span = tracing::debug_span!(
            "reliar.inbox.claim",
            inbox.scope = %scope,
            message.id = %message.id,
            // `r#type`, not `type` — `type` is a Rust keyword; tracing strips the `r#` prefix
            // from the field name, so this still renders as `message.type` (inbox contract §4).
            message.r#type = %message.message_type,
            inbox.outcome = tracing::field::Empty,
            inbox.attempt = tracing::field::Empty,
            inbox.attempts = tracing::field::Empty,
            inbox.record_id = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let result = claim::claim(self, tx, scope, message).await;

            if let Ok(claim) = &result {
                record_claim_outcome(&recording_span, claim);
            }

            result
        }
        .instrument(span)
    }

    /// Marks the row completed in the caller's transaction. Zero rows affected ⇒ `NotClaimed`
    /// (including a row that has since gone dead — the guard is `completed_at IS NULL AND
    /// dead_at IS NULL`).
    // Block form — reason (a): the span's `inbox.scope`/`message.id` fields (inbox contract §4)
    // must be created before the statement they describe runs.
    fn complete(
        &self,
        tx: &mut Transaction<'c, Postgres>,
        scope: &InboxScope,
        id: MessageId,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        let span = tracing::debug_span!(
            "reliar.inbox.complete",
            inbox.scope = %scope,
            message.id = %id,
        );

        async move { outcomes::complete(self, tx, scope, id).await }.instrument(span)
    }

    /// Records a failed attempt on this store's own pool, guarded by `completed_at IS NULL`, and
    /// bounds it at `settings.max_attempts` atomically with the increment (ADR 0042 A.2.4).
    // Block form: `error: &(dyn Error + 'static)` is not `Send`, so its `Display` chain must be
    // extracted into an owned `String` before the async block is built, never inside a plain
    // `async fn` (conventions §3(b); the trait's own rustdoc calls this out) — also reason (a):
    // the `reliar.inbox.fail` span's entry fields must be recorded before the statement runs.
    fn fail(
        &self,
        scope: &InboxScope,
        message: InboxMessage<'_>,
        error: &(dyn std::error::Error + 'static),
    ) -> impl Future<Output = Result<InboxFailure, Self::Error>> + Send {
        let last_error = outcomes::format_error_chain(error);
        let span = tracing::debug_span!(
            "reliar.inbox.fail",
            inbox.scope = %scope,
            message.id = %message.id,
            // `r#type`, not `type` — `type` is a Rust keyword; tracing strips the `r#` prefix
            // from the field name, so this still renders as `message.type` (inbox contract §4).
            message.r#type = %message.message_type,
            inbox.outcome = tracing::field::Empty,
            inbox.attempts = tracing::field::Empty,
            inbox.record_id = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let result = outcomes::fail(self, scope, message, last_error).await;

            if let Ok(failure) = &result {
                record_fail_outcome(&recording_span, failure);
            }

            result
        }
        .instrument(span)
    }

    /// Reads a row for diagnostics. No Reliar code path calls it. No span — the inbox contract's
    /// observability table (§4) does not list `find`, since no Reliar code path calls it.
    async fn find(
        &self,
        scope: &InboxScope,
        id: MessageId,
    ) -> Result<Option<InboxRecord>, Self::Error> {
        purge::find(self, scope, id).await
    }

    /// One bounded pass, three statements, each capped at `request.batch_size`: the
    /// completed-row, incomplete-row and dead-row deletes.
    // Block form — reason (a): the `reliar.inbox.purge` span must exist before the three
    // statements it will report on run.
    fn purge(
        &self,
        request: InboxPurgeRequest,
    ) -> impl Future<Output = Result<InboxPurgeReport, Self::Error>> + Send {
        let span = tracing::debug_span!(
            "reliar.inbox.purge",
            inbox.completed_deleted = tracing::field::Empty,
            inbox.incomplete_deleted = tracing::field::Empty,
            inbox.dead_deleted = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let result = purge::purge(self, request).await;

            if let Ok(report) = &result {
                recording_span.record("inbox.completed_deleted", report.completed_deleted);
                recording_span.record("inbox.incomplete_deleted", report.incomplete_deleted);
                recording_span.record("inbox.dead_deleted", report.dead_deleted);
            }

            result
        }
        .instrument(span)
    }
}

/// The `reliar.inbox.claim` span's outcome fields (ADR 0042 Amendment C.5): `inbox.outcome`
/// always, `inbox.attempt` on `Claimed`, `inbox.record_id` + `inbox.attempts` on `Dead`. Never
/// recorded on an `Err` path — the caller's own log carries the failure.
fn record_claim_outcome(span: &tracing::Span, claim: &InboxClaim) {
    match claim {
        InboxClaim::Claimed { attempt } => {
            span.record("inbox.outcome", "claimed");
            span.record("inbox.attempt", attempt);
        }
        InboxClaim::AlreadyCompleted { .. } => {
            span.record("inbox.outcome", "already_completed");
        }
        InboxClaim::InProgress => {
            span.record("inbox.outcome", "in_progress");
        }
        InboxClaim::Dead { id, attempts, .. } => {
            span.record("inbox.outcome", "dead");
            span.record("inbox.record_id", tracing::field::display(id));
            span.record("inbox.attempts", attempts);
        }
        // `InboxClaim` is `#[non_exhaustive]`; every variant this crate's contract defines is
        // matched above.
        _ => {}
    }
}

/// The `reliar.inbox.fail` span's outcome fields (inbox contract §4): `inbox.outcome` always,
/// `inbox.attempts`/`inbox.record_id` where the variant carries them. Never recorded on an `Err`
/// path.
fn record_fail_outcome(span: &tracing::Span, failure: &InboxFailure) {
    match failure {
        InboxFailure::Recorded { attempts } => {
            span.record("inbox.outcome", "recorded");
            span.record("inbox.attempts", attempts);
        }
        InboxFailure::Dead {
            id,
            attempts,
            dead_at: _,
        } => {
            span.record("inbox.outcome", "dead");
            span.record("inbox.attempts", attempts);
            span.record("inbox.record_id", tracing::field::display(id));
        }
        InboxFailure::AlreadyCompleted => {
            span.record("inbox.outcome", "already_completed");
        }
        // `InboxFailure` is `#[non_exhaustive]`; every variant this crate's contract defines is
        // matched above.
        _ => {}
    }
}