reliar-inbox 0.2.0

//! Transactional deduplication of inbound messages.

use reliar_core::{Classify, MessageId};
use tracing::Instrument as _;

use crate::claim::{InboxClaim, InboxFailure, InboxOutcome};
use crate::error::InboxProcessError;
use crate::handler::InboxHandler;
use crate::message::InboxMessage;
use crate::purge::{InboxPurgeReport, InboxPurgeRequest};
use crate::record::InboxRecord;
use crate::scope::InboxScope;

/// Transactional deduplication of inbound messages, keyed `(scope, message_id)`.
///
/// `Tx` is the provider's transaction type — `sqlx::Transaction<'_, Postgres>` for
/// `reliar-store-postgres`. A **type parameter**, exactly as on `reliar-outbox`'s
/// `OutboxEnqueue<Tx>`, so this crate names no storage type.
///
/// **Two transaction models on purpose.** [`Self::claim`] and [`Self::complete`] run in the
/// caller's transaction — that is the guarantee. [`Self::fail`], [`Self::find`] and
/// [`Self::purge`] run on the provider's own pool: `fail` records the failure of the very
/// transaction that has just been rolled back, so it cannot live in it (ADR 0042 §4).
///
/// [`Self::claim`] then [`Self::complete`] — a **compiled, never-called generic function**
/// (ADR 0043 §7: this crate ships no store to run a doctest against; `reliar-store-postgres`'s
/// own docs carry a runnable example over `PostgresInboxStore`):
///
/// ```
/// # use reliar_inbox::{InboxClaim, InboxMessage, InboxScope, InboxStore};
/// #
/// async fn claim_then_complete<Tx, S: InboxStore<Tx>>(
///     store: &S,
///     tx: &mut Tx,
///     scope: &InboxScope,
///     message: InboxMessage<'_>,
/// ) {
///     if let Ok(InboxClaim::Claimed { .. }) = store.claim(tx, scope, message).await {
///         let _ = store.complete(tx, scope, message.id).await;
///     }
/// }
/// ```
pub trait InboxStore<Tx>: Send + Sync {
    /// What inbox operations fail with. [`Classify`] so a caller can log a permanent failure
    /// differently from a transient one without a downcast.
    type Error: std::error::Error + Send + Sync + 'static + Classify;

    /// Claims `message` for `scope` in the caller's transaction — see [`InboxClaim`] for the
    /// four answers and what the caller owes each of them.
    ///
    /// Call it as the **first** statement of the transaction. `Ok` never leaves `tx` unusable;
    /// an `Err` may, and whether it does is the provider's contract (with
    /// `reliar-store-postgres` it does — PostgreSQL aborts the transaction).
    ///
    /// Issues no network I/O beyond its own statements, and never commits, rolls back or
    /// otherwise consumes `tx`.
    ///
    /// Non-generic on purpose: a `claim<T>(…, &Envelope<T>)` would monomorphize the provider's
    /// SQL path per body type (ADR 0042 A.2.2).
    ///
    /// # Errors
    ///
    /// Provider-defined. Treat any `Err` as *abort this transaction*.
    ///
    /// ```
    /// # use reliar_inbox::{InboxClaim, InboxMessage, InboxScope, InboxStore};
    /// #
    /// async fn claim_only<Tx, S: InboxStore<Tx>>(
    ///     store: &S,
    ///     tx: &mut Tx,
    ///     scope: &InboxScope,
    ///     message: InboxMessage<'_>,
    /// ) -> Option<InboxClaim> {
    ///     store.claim(tx, scope, message).await.ok()
    /// }
    /// ```
    fn claim(
        &self,
        tx: &mut Tx,
        scope: &InboxScope,
        message: InboxMessage<'_>,
    ) -> impl Future<Output = Result<InboxClaim, Self::Error>> + Send;

    /// Marks the row completed in the caller's transaction, at database time. The last
    /// statement before the caller's `commit`.
    ///
    /// Preserves `last_error` from earlier failed attempts — a message that eventually
    /// succeeded keeps the evidence of why it did not the first time.
    ///
    /// Its guard is `completed_at IS NULL AND dead_at IS NULL` — the second clause is not
    /// cosmetic: with `ck_inbox_terminal` in place, completing a dead row would trip the check
    /// constraint and surface a raw database error where this contract promises a clean
    /// "no claimed row".
    ///
    /// **A caller that commits without calling this** — by calling [`Self::claim`] directly and
    /// skipping `complete`, or by committing after `complete` itself returned an error — leaves a
    /// committed, uncompleted row. It is then indistinguishable from a row [`Self::fail`]
    /// created: the next redelivery's `claim` answers `Claimed { attempt: 1 }` again and the
    /// handler re-runs over already-committed business writes. [`Self::process`] never risks
    /// this (it always calls `complete` before returning `Processed`); a caller that drives
    /// `claim`/`complete` itself must call `complete` before its own commit to keep the guarantee.
    ///
    /// Keeps `id: MessageId` — completion writes no new columns, so it needs no [`InboxMessage`]
    /// view.
    ///
    /// # Errors
    ///
    /// Provider-defined, plus a provider error for "no claimed row" — reachable by misuse
    /// (completing without a `Claimed`, or after the transaction aborted), by completing a row
    /// that has since gone dead, and, more benignly, by a concurrent retention [`Self::purge`]
    /// deleting the row `claim` adopted in between: the result is a spurious `Err` here, a
    /// rollback, and a clean redelivery — never a lost or duplicated effect.
    ///
    /// ```
    /// # use reliar_core::MessageId;
    /// # use reliar_inbox::{InboxScope, InboxStore};
    /// #
    /// async fn complete_only<Tx, S: InboxStore<Tx>>(
    ///     store: &S,
    ///     tx: &mut Tx,
    ///     scope: &InboxScope,
    ///     id: MessageId,
    /// ) {
    ///     let _ = store.complete(tx, scope, id).await;
    /// }
    /// ```
    fn complete(
        &self,
        tx: &mut Tx,
        scope: &InboxScope,
        id: MessageId,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;

    /// Records a failed attempt **on the provider's own pool**, in its own short transaction.
    /// Call it *after* rolling the handler's transaction back.
    ///
    /// Increments `attempts` and stores `error`'s `Display` chain (truncated to 2 KiB at a char
    /// boundary with a `"…[truncated]"` marker, like [`crate::InboxRecord::last_error`]),
    /// creating the row if the rollback removed it. A row that is already completed is left
    /// untouched: a stale attempt cannot un-complete work another consumer finished.
    ///
    /// Takes the same [`InboxMessage`] view [`Self::claim`] does, because it **creates** the row
    /// when the rollback removed it and must supply every `NOT NULL` column — and it returns an
    /// [`InboxFailure`], because whether the row just went dead decides the caller's next broker
    /// call.
    ///
    /// An implementation SHALL bound recorded failures at a configured `max_attempts` and SHALL
    /// apply the transition to `dead_at` **atomically with the increment** — reading the count
    /// and writing the transition back separately races two concurrent `fail`s and can skip the
    /// transition or apply it twice.
    ///
    /// **Best-effort bookkeeping.** Skipping it — or crashing before it — costs an uncounted
    /// attempt and nothing else; no Reliar decision is taken on the count. `attempts` therefore
    /// stays a **lower** bound, so `max_attempts` bounds *recorded* failures only, and nothing
    /// about the row bounds effects **outside** the database.
    ///
    /// **Implementors:** `error: &dyn Error` is not `Send` (`&T: Send` requires `T: Sync`, and
    /// `dyn Error` is not `Sync`), so a plain `async fn` that carries it across the async block's
    /// construction will not satisfy this method's `+ Send` return bound. Extract `error`'s
    /// `Display` chain into an owned `String` synchronously, before the async block is built —
    /// see `reliar-store-postgres`'s `PostgresInboxStore::fail` for the reference shape.
    /// `InboxMessage<'_>` itself may be captured: `MessageType` and `CorrelationId` are both
    /// `Sync`.
    ///
    /// # Errors
    ///
    /// Provider-defined. An `Err` here changes nothing about the message's fate; log it and
    /// `nak`.
    ///
    /// ```
    /// # use reliar_inbox::{InboxFailure, InboxMessage, InboxScope, InboxStore};
    /// #
    /// # #[derive(Debug)]
    /// # struct Boom;
    /// # impl std::fmt::Display for Boom {
    /// #     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    /// #         f.write_str("boom")
    /// #     }
    /// # }
    /// # impl std::error::Error for Boom {}
    /// #
    /// async fn fail_only<Tx, S: InboxStore<Tx>>(
    ///     store: &S,
    ///     scope: &InboxScope,
    ///     message: InboxMessage<'_>,
    /// ) -> Option<InboxFailure> {
    ///     store.fail(scope, message, &Boom).await.ok()
    /// }
    /// ```
    fn fail(
        &self,
        scope: &InboxScope,
        message: InboxMessage<'_>,
        error: &(dyn std::error::Error + 'static),
    ) -> impl Future<Output = Result<InboxFailure, Self::Error>> + Send;

    /// Reads a row for diagnostics. No Reliar code path calls it.
    ///
    /// # Errors
    ///
    /// Provider-defined.
    ///
    /// ```
    /// # use reliar_core::MessageId;
    /// # use reliar_inbox::{InboxRecord, InboxScope, InboxStore};
    /// #
    /// async fn find_only<Tx, S: InboxStore<Tx>>(
    ///     store: &S,
    ///     scope: &InboxScope,
    ///     id: MessageId,
    /// ) -> Option<InboxRecord> {
    ///     store.find(scope, id).await.ok().flatten()
    /// }
    /// ```
    fn find(
        &self,
        scope: &InboxScope,
        id: MessageId,
    ) -> impl Future<Output = Result<Option<InboxRecord>, Self::Error>> + Send;

    /// Deletes rows past retention, bounded by `request.batch_size`, on the provider's own pool.
    /// Idempotent and safe to run concurrently with consumers. `Tx` is inert here.
    ///
    /// **Retention is the redelivery window, not a storage budget** — see [`InboxPurgeRequest`].
    ///
    /// # Errors
    ///
    /// Provider-defined.
    ///
    /// ```
    /// # use reliar_inbox::{InboxPurgeReport, InboxPurgeRequest, InboxStore};
    /// #
    /// async fn purge_only<Tx, S: InboxStore<Tx>>(store: &S) -> Option<InboxPurgeReport> {
    ///     store.purge(InboxPurgeRequest::default()).await.ok()
    /// }
    /// ```
    fn purge(
        &self,
        request: InboxPurgeRequest,
    ) -> impl Future<Output = Result<InboxPurgeReport, Self::Error>> + Send;

    /// The happy path in one call: claim, branch, run `handler`, complete.
    ///
    /// **Never commits and never calls [`Self::fail`]** — it holds only `&mut Tx`, and `fail`
    /// needs a different connection than the transaction being rolled back. The caller owns
    /// both, and the ordering is the part that must not be improvised — **six** branches, and
    /// the `Err(Handler)` row branches again on what `fail` returned:
    ///
    /// | Result | Caller does |
    /// |---|---|
    /// | `Ok(Processed(v))` | `tx.commit()`, **then** ack |
    /// | `Ok(AlreadyCompleted { .. })` | `tx.rollback()`, ack |
    /// | `Ok(InProgress)` | `tx.rollback()`, `nak` with a delay |
    /// | `Ok(Dead { id, .. })` | `tx.rollback()`, log `id`, **`term`** — never re-run the handler |
    /// | `Err(Handler(e))` | `tx.rollback()`, then `self.fail(scope, message, &e)` and follow its [`InboxFailure`]: `Recorded` ⇒ `nak`, `Dead` ⇒ `term`, `AlreadyCompleted` ⇒ ack |
    /// | `Err(Store(e))` | `tx.rollback()`, log, `nak` |
    ///
    /// A host that ignores `Dead` will `nak` forever and the message will bounce until the
    /// broker's own `max_deliver`.
    ///
    /// Acking before the commit succeeds turns at-least-once into at-most-once: the message is
    /// gone and its effects were rolled back. Following this table keeps [`Self::complete`]'s own
    /// "commit without complete" duplicate window from ever opening: `process` calls `complete`
    /// on every path that reaches `tx.commit()`, so a caller that only ever commits on
    /// `Ok(Processed(_))` never needs `complete`'s rustdoc to protect itself.
    ///
    /// Implementors **SHALL NOT** override this method — it is a fixed spelling of
    /// `claim`/`complete`, not an extension point.
    ///
    /// # Errors
    ///
    /// [`InboxProcessError::Handler`] when `handler` fails, [`InboxProcessError::Store`] for any
    /// claim or completion failure.
    ///
    /// ```
    /// # use reliar_inbox::{InboxHandler, InboxMessage, InboxOutcome, InboxProcessError, InboxScope, InboxStore};
    /// #
    /// # struct RecordOrder;
    /// # impl<Tx: Send> InboxHandler<Tx> for RecordOrder {
    /// #     type Output = &'static str;
    /// #     type Error = std::convert::Infallible;
    /// #     async fn handle(&self, _tx: &mut Tx) -> Result<Self::Output, Self::Error> {
    /// #         Ok("order recorded")
    /// #     }
    /// # }
    /// #
    /// async fn process_only<Tx: Send, S: InboxStore<Tx>>(
    ///     store: &S,
    ///     tx: &mut Tx,
    ///     scope: &InboxScope,
    ///     message: InboxMessage<'_>,
    /// ) -> Result<InboxOutcome<&'static str>, InboxProcessError<S::Error, std::convert::Infallible>> {
    ///     store.process(tx, scope, message, &RecordOrder).await
    /// }
    /// ```
    // Block form: a provided method with an `impl Future` signature must use it (conventions §3;
    // also reason (a) — the span below is opened eagerly, at call time, not on first poll).
    #[allow(
        clippy::type_complexity,
        reason = "the return type names exactly the two outcomes `process` can produce; a type \
                  alias would need `Self::Error` and `H::Error` as parameters and read no clearer"
    )]
    fn process<H>(
        &self,
        tx: &mut Tx,
        scope: &InboxScope,
        message: InboxMessage<'_>,
        handler: &H,
    ) -> impl Future<
        Output = Result<InboxOutcome<H::Output>, InboxProcessError<Self::Error, H::Error>>,
    > + Send
    where
        H: InboxHandler<Tx> + Sync,
        Tx: Send,
    {
        let span = tracing::info_span!(
            "reliar.inbox.process",
            inbox.scope = %scope,
            message.id = %message.id,
            message.r#type = %message.message_type,
            inbox.outcome = tracing::field::Empty,
            inbox.record_id = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let outcome = match self
                .claim(tx, scope, message)
                .await
                .map_err(InboxProcessError::Store)?
            {
                InboxClaim::AlreadyCompleted { completed_at } => {
                    InboxOutcome::AlreadyCompleted { completed_at }
                }

                InboxClaim::InProgress => InboxOutcome::InProgress,

                InboxClaim::Dead {
                    id,
                    attempts,
                    dead_at,
                } => InboxOutcome::Dead {
                    id,
                    attempts,
                    dead_at,
                },

                InboxClaim::Claimed { .. } => {
                    let output = handler
                        .handle(tx)
                        .await
                        .map_err(InboxProcessError::Handler)?;

                    self.complete(tx, scope, message.id)
                        .await
                        .map_err(InboxProcessError::Store)?;

                    InboxOutcome::Processed(output)
                }
            };

            recording_span.record("inbox.outcome", outcome_label(&outcome));

            if let InboxOutcome::Dead { id, .. } = &outcome {
                recording_span.record("inbox.record_id", tracing::field::display(id));
            }

            Ok(outcome)
        }
        .instrument(span)
    }
}

/// The `inbox.outcome` span field value for [`InboxStore::process`]'s success path. Never
/// recorded on an error path — the field stays empty, and the caller's own log carries the
/// failure.
fn outcome_label<T>(outcome: &InboxOutcome<T>) -> &'static str {
    match outcome {
        InboxOutcome::Processed(_) => "processed",
        InboxOutcome::AlreadyCompleted { .. } => "already_completed",
        InboxOutcome::InProgress => "in_progress",
        InboxOutcome::Dead { .. } => "dead",
    }
}