reliar_store_postgres/outbox/

outbox_store.rs

//! The [`PostgresOutboxStore`] type itself: fields, construction (`new`/`with_settings`/
//! `with_serializer`), the small shared helper every other store module calls back into
//! (`content_type`), and the whole `OutboxStore` trait impl (one file per public trait,
//! `docs/architecture/store-postgres-layout.md` Part II §6): each method's body calls straight into its concern-layer module
//! (`claim`, `outcomes`, `purge`). The private helpers below the impl block (`fenced_ids`,
//! `FailBatches`/`classify_failures`/`apply_fail_batches`, `to_millis`) are shared across more
//! than one of those method bodies, so they stay out of the impl block itself.

use std::sync::Arc;
use std::time::Duration;

use reliar_core::{ContentType, Serializer};
use reliar_outbox::{
    AcquireRequest, AcquiredBatch, FailedRecord, FailureOutcome, OutboxStats, OutboxStore,
    PoisonedRow, PurgeReport, PurgeRequest, RecordRef, WorkerId,
};
use sqlx::{PgConnection, PgPool};

use crate::connection::session::Session;
use crate::records::{RawRow, decode_row};
use crate::settings::PostgresOutboxSettings;

#[cfg(feature = "json")]
use reliar_core::JsonSerializer;

use super::claim as claim_repo;
use super::error::PostgresOutboxError;
use super::outcomes as outcomes_repo;
use super::purge as purge_repo;

/// Reliar's PostgreSQL outbox provider. Cheap to clone into an `AppState` — it wraps a
/// [`PgPool`]; no outer `Arc` required. The connection pool stays the host's: Reliar never owns
/// or reads a `DATABASE_URL`.
///
/// The default type parameter only exists behind the crate's default `json` feature: under
/// `--no-default-features` there is no default, so [`Self::with_serializer`] is the only
/// constructor and `cargo hack --feature-powerset` compiles every combination. This block's
/// `PostgresOutboxStore::new` leans on that default, so it only compiles under `json`; 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() -> Result<(), Box<dyn std::error::Error>> {
/// use reliar_store_postgres::{PostgresOutboxStore, migrate};
/// use sqlx::postgres::PgPoolOptions;
///
/// let pool = PgPoolOptions::new()
///     .connect(&std::env::var("DATABASE_URL")?)
///     .await?;
/// migrate(&pool, Default::default()).await?;
///
/// let store = PostgresOutboxStore::new(pool);
/// // `store` now implements `OutboxEnqueue`, `OutboxStore` and `OutboxDeadLetters` —
/// // hand it to an application's write path and to an `OutboxDispatcher`.
/// # Ok(())
/// # }
/// ```
#[non_exhaustive]
pub struct PostgresOutboxStore<
    #[cfg(feature = "json")] Ser = JsonSerializer,
    #[cfg(not(feature = "json"))] Ser,
> {
    // `pub(super)`: `outbox_store_dead_letters.rs` reads this directly — its whole
    // `OutboxDeadLetters` impl (signature and body) lives there rather than delegating from this
    // file, unlike `claim`/`outcomes`/`purge`, which are read as `&self.session` from the trait
    // methods below. `outbox_store_enqueue.rs` never touches it: `OutboxEnqueue` runs inside the
    // caller's own transaction, not `Session::run`.
    pub(super) session: Session,

    // `pub(super)`: `outbox_store_enqueue.rs` reads this directly — its whole `OutboxEnqueue` impl
    // lives there, the same reason `outbox_store_dead_letters.rs` reads `session` above.
    pub(super) serializer: Arc<Ser>,
}

/// **Manual impl, never derived**: a derived `Clone` would condition on `Ser: Clone`. The
/// serializer is held as `Arc<Ser>` — stateless and cheap to share — so cloning the store never
/// requires the serializer itself to be `Clone`.
impl<Ser> Clone for PostgresOutboxStore<Ser> {
    fn clone(&self) -> Self {
        Self {
            session: self.session.clone(),
            serializer: Arc::clone(&self.serializer),
        }
    }
}

impl<Ser> std::fmt::Debug for PostgresOutboxStore<Ser> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PostgresOutboxStore")
            .field("session", &self.session)
            .finish_non_exhaustive()
    }
}

impl<Ser: Serializer + Send + Sync + 'static> PostgresOutboxStore<Ser> {
    /// Wraps `pool` with `settings` and `serializer`. Performs **no I/O**: it issues no query,
    /// opens no connection and verifies nothing about the database. The pool stays the host's.
    ///
    /// Call [`crate::migrate`] (or apply the published SQL through your own pipeline) **before**
    /// the first store call, and make sure the connection's `search_path` resolves the
    /// unqualified name `outbox` to the migrated schema — see the crate docs. An un-migrated or
    /// unreachable table surfaces at the first statement as
    /// [`PostgresOutboxError::NotMigrated`], never here.
    ///
    /// This example uses [`reliar_core::JsonSerializer`], 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() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_core::JsonSerializer;
    /// use reliar_store_postgres::{PostgresOutboxSettings, PostgresOutboxStore};
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store =
    ///     PostgresOutboxStore::with_serializer(pool, PostgresOutboxSettings::default(), JsonSerializer);
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    #[allow(
        clippy::needless_pass_by_value,
        reason = "the public signature takes settings by value (ADR 0047 §1); only \
                  statement_timeout is read today, but PostgresOutboxSettings is #[non_exhaustive] \
                  and may grow a field this constructor needs to own or move out of later"
    )]
    pub fn with_serializer(
        pool: PgPool,
        settings: PostgresOutboxSettings,
        serializer: Ser,
    ) -> Self {
        let session = Session::new(pool, settings.statement_timeout);

        Self {
            session,
            serializer: Arc::new(serializer),
        }
    }

    /// The `ContentType` this store writes to every row — `Serializer::content_type()`. The
    /// only way a caller can predict the `content_type` of an envelope it will later acquire:
    /// `enqueue` writes this value, ignoring whatever `envelope.metadata.delivery.content_type`
    /// held. `PostgresOutboxStore::new` here leans on the 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(pool: sqlx::PgPool) -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::PostgresOutboxStore;
    ///
    /// let store = PostgresOutboxStore::new(pool);
    /// assert_eq!(store.content_type().as_str(), "application/json");
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn content_type(&self) -> &ContentType {
        self.serializer.content_type()
    }
}

#[cfg(feature = "json")]
#[cfg_attr(docsrs, doc(cfg(feature = "json")))]
impl PostgresOutboxStore<JsonSerializer> {
    /// Convenience over [`Self::with_serializer`] with [`reliar_core::JsonSerializer`] and
    /// [`PostgresOutboxSettings::default`], behind the crate's default `json` feature. Performs
    /// **no I/O** — see [`Self::with_serializer`].
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::PostgresOutboxStore;
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresOutboxStore::new(pool);
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self::with_serializer(pool, PostgresOutboxSettings::default(), JsonSerializer)
    }

    /// Convenience over [`Self::with_serializer`] with [`reliar_core::JsonSerializer`] and
    /// explicit `settings`, behind the crate's default `json` feature. Performs **no I/O** — see
    /// [`Self::with_serializer`].
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::{PostgresOutboxSettings, PostgresOutboxStore};
    /// use sqlx::postgres::PgPoolOptions;
    /// use std::time::Duration;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresOutboxStore::with_settings(
    ///     pool,
    ///     PostgresOutboxSettings::default().statement_timeout(Duration::from_secs(2)),
    /// );
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn with_settings(pool: PgPool, settings: PostgresOutboxSettings) -> Self {
        Self::with_serializer(pool, settings, JsonSerializer)
    }
}

impl<Ser: Serializer + Send + Sync + 'static> OutboxStore for PostgresOutboxStore<Ser> {
    type Error = PostgresOutboxError;

    /// The canonical single-statement claim (ADR 0006): a CTE
    /// `SELECT … FOR UPDATE SKIP LOCKED` feeding an `UPDATE … RETURNING`, so the row lock is
    /// released before this future resolves and no network I/O to a publisher can ever happen
    /// while it is held.
    ///
    /// A row this call cannot decode is **excluded from `records`**, reported in `poisoned`,
    /// and **moved to dead** with `DeadReason::Undecodable` by a follow-up statement fenced by the
    /// claim token this claim just stamped (ADR 0046 Amendment A) — the batch continues rather
    /// than failing outright (ADR 0008).
    async fn acquire(&self, request: AcquireRequest) -> Result<AcquiredBatch, Self::Error> {
        let session = &self.session;
        let batch_size = i64::from(request.batch_size);
        let lease_ms = i64::try_from(request.lease.as_millis()).unwrap_or(i64::MAX);
        let worker = request.worker.as_str();

        let rows: Vec<RawRow> = session
            .run(async |conn: &mut PgConnection| {
                claim_repo::claim_rows(
                    &mut *conn,
                    claim_repo::ClaimRowsParams {
                        batch_size,
                        worker,
                        lease_ms,
                    },
                )
                .await
            })
            .await
            .map_err(|e| session.map_err::<PostgresOutboxError>(e))?;

        let mut records = Vec::with_capacity(rows.len());
        let mut poisoned = Vec::new();
        let mut poisoned_ids = Vec::new();
        let mut poisoned_tokens = Vec::new();
        let mut poisoned_errors = Vec::new();

        for raw in rows {
            // Captured before `decode_row` consumes `raw` by value — the poison sweep below must
            // fence on the token this very claim just stamped, and `RowError` (what a decode
            // failure returns) carries no such column (ADR 0046 Amendment A.5 item 9).
            let claim_token = raw.claim_token;

            match decode_row(raw) {
                Ok(record) => records.push(record),
                Err(err) => {
                    poisoned_ids.push(err.id.as_uuid());
                    poisoned_tokens.push(claim_token);
                    poisoned_errors.push(crate::records::truncate_last_error(err.detail.clone()));

                    poisoned.push(PoisonedRow::new(err.id, err.message_id, err.detail));
                }
            }
        }

        if !poisoned_ids.is_empty() {
            // Not an observed publish attempt, so `attempts` is untouched (ADR 0009: `attempts`
            // counts outcomes, never claims) — only the lease clears and the row goes dead. Runs
            // under the same `Session::run` policy as the claim itself, so a slow poison sweep
            // stays bounded by a non-zero `statement_timeout` too.
            //
            // **Best-effort (ADR 0039 §4): a sweep failure never turns a committed claim into an
            // `Err`.** The claim above has already committed and its rows are already leased to
            // this caller; failing the whole batch here would strand the N healthy rows for a
            // full lease over a problem with the poisoned ones. On failure this only logs — the
            // poisoned rows keep their lease and are re-attempted (sweep or publish) once it
            // lapses, so `poisoned` means "could not decode and an attempt was made to deaden",
            // not "is dead".
            let undecodable =
                crate::records::encode_dead_reason(reliar_outbox::DeadReason::Undecodable);
            let sweep_result = session
                .run(async |conn: &mut PgConnection| {
                    claim_repo::poison_sweep_rows(
                        &mut *conn,
                        claim_repo::PoisonSweepRowsParams {
                            ids: &poisoned_ids,
                            tokens: &poisoned_tokens,
                            errors: &poisoned_errors,
                            dead_reason: undecodable,
                        },
                    )
                    .await
                })
                .await;

            if let Err(err) = sweep_result {
                // Plain snake_case fields, not the usual dotted `worker.id`/`poisoned.count`
                // convention: `tracing`'s event macro hits a `macro_rules!` parsing
                // ambiguity ("multiple parsing options: built-in NTs tt ('field') or 1 other
                // option") when an explicit `target:` is followed by a dotted field path — a
                // `tracing` macro limitation, not a style choice.
                tracing::warn!(
                    target: "reliar.outbox.acquire",
                    worker_id = %worker,
                    poisoned_count = poisoned_ids.len(),
                    error = %session.map_err::<PostgresOutboxError>(err),
                    "poison sweep failed; the claimed batch is returned and the undecodable rows \
                     stay leased until their lease lapses"
                );
            }
        }

        Ok(AcquiredBatch::new(records, poisoned))
    }

    /// Marks rows published, fenced by each item's claim token (ADR 0046 Amendment A). A row
    /// already completed or reclaimed under a fresh token — by any worker, including this one —
    /// contributes nothing to the count; a shortfall is logged at `warn`, naming the fenced ids,
    /// never an error (ADR 0008, ADR 0046 §5).
    async fn complete(&self, worker: &WorkerId, items: &[RecordRef]) -> Result<u64, Self::Error> {
        let session = &self.session;

        if items.is_empty() {
            return Ok(0);
        }

        let ids: Vec<uuid::Uuid> = items.iter().map(|i| i.id.as_uuid()).collect();
        let tokens: Vec<Option<uuid::Uuid>> = items
            .iter()
            .map(|i| i.claim_token.map(|t| t.as_uuid()))
            .collect();
        let applied = session
            .run(async |conn: &mut PgConnection| {
                outcomes_repo::complete_rows(
                    &mut *conn,
                    outcomes_repo::CompleteRowsParams {
                        ids: &ids,
                        tokens: &tokens,
                    },
                )
                .await
            })
            .await
            .map_err(|e| session.map_err::<PostgresOutboxError>(e))?;

        if applied.len() < ids.len() {
            tracing::warn!(
                target: "reliar.outbox.complete",
                requested = ids.len(),
                worker.id = %worker,
                applied = applied.len(),
                fenced_ids = ?fenced_ids(&ids, &applied),
                "fewer rows completed than requested — the fenced rows belong to a superseded claim"
            );
        }

        Ok(applied.len() as u64)
    }

    /// Applies each item's [`FailureOutcome`](reliar_outbox::FailureOutcome), fenced by each
    /// item's claim token. Retry rows get `available_at = now() + delay` computed in SQL
    /// (ADR 0009); dead rows get `dead_at`/`dead_reason` set together (`ck_outbox_dead_reason`).
    /// Both increment `attempts` — on outcome, never on claim.
    async fn fail(&self, worker: &WorkerId, items: &[FailedRecord]) -> Result<u64, Self::Error> {
        let session = &self.session;

        if items.is_empty() {
            return Ok(0);
        }

        let batches = classify_failures(items);
        let requested = batches.retry_ids.len() + batches.dead_ids.len();
        let (applied_retry, applied_dead) = apply_fail_batches(session, &batches).await?;
        let applied = applied_retry.len() + applied_dead.len();

        if applied < requested {
            let ids: Vec<uuid::Uuid> = batches
                .retry_ids
                .iter()
                .chain(batches.dead_ids.iter())
                .copied()
                .collect();
            let applied_ids: Vec<uuid::Uuid> = applied_retry
                .iter()
                .chain(applied_dead.iter())
                .copied()
                .collect();

            tracing::warn!(
                target: "reliar.outbox.fail",
                requested,
                worker.id = %worker,
                applied,
                fenced_ids = ?fenced_ids(&ids, &applied_ids),
                "fewer rows failed than requested — the fenced rows belong to a superseded claim"
            );
        }

        Ok(applied as u64)
    }

    /// Clears the lease for rows whose claim token still matches. `available_at` and `attempts`
    /// are untouched — a release is not a failure.
    async fn release(&self, worker: &WorkerId, items: &[RecordRef]) -> Result<u64, Self::Error> {
        let session = &self.session;

        if items.is_empty() {
            return Ok(0);
        }

        let ids: Vec<uuid::Uuid> = items.iter().map(|i| i.id.as_uuid()).collect();
        let tokens: Vec<Option<uuid::Uuid>> = items
            .iter()
            .map(|i| i.claim_token.map(|t| t.as_uuid()))
            .collect();
        let applied = session
            .run(async |conn: &mut PgConnection| {
                outcomes_repo::release_rows(
                    &mut *conn,
                    outcomes_repo::ReleaseRowsParams {
                        ids: &ids,
                        tokens: &tokens,
                    },
                )
                .await
            })
            .await
            .map_err(|e| session.map_err::<PostgresOutboxError>(e))?;

        if applied.len() < ids.len() {
            tracing::warn!(
                target: "reliar.outbox.release",
                requested = ids.len(),
                worker.id = %worker,
                applied = applied.len(),
                fenced_ids = ?fenced_ids(&ids, &applied),
                "fewer rows released than requested — the fenced rows belong to a superseded claim"
            );
        }

        Ok(applied.len() as u64)
    }

    /// Renews the lease by moving `available_at` to `now() + lease` for rows whose claim token
    /// still matches, without rotating it (ADR 0046 Amendment A). `available_at` is the lease
    /// clock, and the only one (ADR 0050 §1). Best-effort: a shortfall means the claim was
    /// superseded.
    async fn extend_lease(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
        lease: Duration,
    ) -> Result<u64, Self::Error> {
        let session = &self.session;

        if items.is_empty() {
            return Ok(0);
        }

        let ids: Vec<uuid::Uuid> = items.iter().map(|i| i.id.as_uuid()).collect();
        let tokens: Vec<Option<uuid::Uuid>> = items
            .iter()
            .map(|i| i.claim_token.map(|t| t.as_uuid()))
            .collect();
        let lease_ms = i64::try_from(lease.as_millis()).unwrap_or(i64::MAX);
        let applied = session
            .run(async |conn: &mut PgConnection| {
                outcomes_repo::extend_lease_rows(
                    &mut *conn,
                    outcomes_repo::ExtendLeaseRowsParams {
                        ids: &ids,
                        tokens: &tokens,
                        lease_ms,
                    },
                )
                .await
            })
            .await
            .map_err(|e| session.map_err::<PostgresOutboxError>(e))?;

        if applied.len() < ids.len() {
            tracing::warn!(
                target: "reliar.outbox.extend_lease",
                requested = ids.len(),
                worker.id = %worker,
                applied = applied.len(),
                fenced_ids = ?fenced_ids(&ids, &applied),
                "fewer leases renewed than requested — the fenced rows belong to a superseded claim"
            );
        }

        Ok(applied.len() as u64)
    }

    /// **One bounded pass, three statements, each capped at `request.batch_size`**:
    /// published-row delete, dead-row delete, and the expired→dead sweep — none of the
    /// three is ever an unbounded `DELETE`/`UPDATE`. The sweep's predicate carries the
    /// not-currently-leased guard (`locked_by IS NULL OR available_at <= now()`, ADR 0050 §2.3),
    /// so it never transitions a row a live worker still owns — that worker's own
    /// `complete`/`fail` wins, and the row becomes sweepable only once its lease lapses.
    async fn purge(&self, request: PurgeRequest) -> Result<PurgeReport, Self::Error> {
        let session = &self.session;
        let batch_size = i64::from(request.batch_size);
        let expired_reason = crate::records::encode_dead_reason(reliar_outbox::DeadReason::Expired);
        let published_ms = request.published_retention.map(to_millis);
        let dead_ms = request.dead_retention.map(to_millis);

        let (published_deleted, dead_deleted, expired_to_dead) = session
            .run(async |conn: &mut PgConnection| {
                let published_deleted = match published_ms {
                    Some(retention_ms) => {
                        purge_repo::purge_published_rows(
                            &mut *conn,
                            purge_repo::PurgePublishedRowsParams {
                                retention_ms,
                                batch_size,
                            },
                        )
                        .await?
                    }
                    None => 0,
                };

                let dead_deleted = match dead_ms {
                    Some(retention_ms) => {
                        purge_repo::purge_dead_retention_rows(
                            &mut *conn,
                            purge_repo::PurgeDeadRetentionRowsParams {
                                retention_ms,
                                batch_size,
                            },
                        )
                        .await?
                    }
                    None => 0,
                };

                let expired_to_dead = purge_repo::purge_expired_sweep_rows(
                    &mut *conn,
                    purge_repo::PurgeExpiredSweepRowsParams {
                        batch_size,
                        dead_reason: expired_reason,
                    },
                )
                .await?;

                Ok((published_deleted, dead_deleted, expired_to_dead))
            })
            .await
            .map_err(|e| session.map_err::<PostgresOutboxError>(e))?;

        Ok(PurgeReport::new(
            published_deleted,
            dead_deleted,
            expired_to_dead,
        ))
    }

    /// One statement, **four independently planned scalar subqueries** (ADR 0040 §3; supersedes
    /// the earlier single-scan `FILTER`-aggregate form, which was `O(table)`). Each subquery is
    /// aimed at its own partial index — `pending` and `oldest_pending_available_at` at
    /// `ix_outbox_claimable` (an index-only scan can evaluate a filter on its `INCLUDE`d
    /// `expires_at`), `dead` at `ix_outbox_dead_cursor`, `expired_pending` at
    /// `ix_outbox_expires` — so the cost is `O(claimable backlog)`/`O(dead rows)`/`O(expired
    /// rows)`, never `O(table)`, and `oldest_pending_available_at` is a single-row `LIMIT`. One
    /// round trip, one transaction snapshot (`now()` evaluated once), so `as_of` and the four
    /// values are consistent with each other even though each is planned separately. Measured at
    /// 100k rows (mixed pending/leased/published/dead/expired) on a vacuumed table, every
    /// subquery plans as an index-only scan with zero heap fetches.
    async fn stats(&self) -> Result<OutboxStats, Self::Error> {
        let session = &self.session;
        let row = session
            .run(async |conn: &mut PgConnection| purge_repo::stats_row(&mut *conn).await)
            .await
            .map_err(|e| session.map_err::<PostgresOutboxError>(e))?;

        Ok(OutboxStats::new(
            u64::try_from(row.pending).unwrap_or(0),
            u64::try_from(row.dead).unwrap_or(0),
            u64::try_from(row.expired_pending).unwrap_or(0),
            row.oldest_pending_available_at,
            row.as_of,
        ))
    }
}

/// The ids in `requested` that a guarded write did not touch — a fresher claim now owns them, or
/// the reference never carried a live claim's token at all. Allocates only when there is a
/// shortfall to report. Shared by `complete`/`fail`/`release`/`extend_lease` above.
///
/// A caller that passes the same `(id, token)` pair twice in one batch is a caller bug, not a
/// guard failure: `UNNEST`'s join matches the row once per array element, so `RETURNING o.id`
/// (and therefore `applied.len()`) can legitimately exceed the row's own single-update count,
/// which would misreport as a spurious shortfall here rather than a clean over-count.
fn fenced_ids(requested: &[uuid::Uuid], applied: &[uuid::Uuid]) -> Vec<uuid::Uuid> {
    let applied: std::collections::HashSet<&uuid::Uuid> = applied.iter().collect();

    requested
        .iter()
        .filter(|id| !applied.contains(id))
        .copied()
        .collect()
}

/// `fail`'s items, split by [`FailureOutcome`] into the two statement shapes `fail_retry_rows`/
/// `fail_dead_rows` take — grouping this way, rather than one item at a time, keeps `fail` a
/// single pair of statements regardless of how many items of each outcome it was handed.
struct FailBatches {
    retry_ids: Vec<uuid::Uuid>,

    retry_tokens: Vec<Option<uuid::Uuid>>,

    retry_errors: Vec<String>,

    retry_delays: Vec<i64>,

    dead_ids: Vec<uuid::Uuid>,

    dead_tokens: Vec<Option<uuid::Uuid>>,

    dead_errors: Vec<String>,

    dead_reasons: Vec<&'static str>,
}

fn classify_failures(items: &[FailedRecord]) -> FailBatches {
    let mut batches = FailBatches {
        retry_ids: Vec::new(),
        retry_tokens: Vec::new(),
        retry_errors: Vec::new(),
        retry_delays: Vec::new(),
        dead_ids: Vec::new(),
        dead_tokens: Vec::new(),
        dead_errors: Vec::new(),
        dead_reasons: Vec::new(),
    };

    for item in items {
        match item.outcome {
            FailureOutcome::Retry { delay } => {
                batches.retry_ids.push(item.record.id.as_uuid());
                batches
                    .retry_tokens
                    .push(item.record.claim_token.map(|t| t.as_uuid()));
                batches.retry_errors.push(item.error.clone());

                batches
                    .retry_delays
                    .push(i64::try_from(delay.as_millis()).unwrap_or(i64::MAX));
            }
            FailureOutcome::Dead { reason } => {
                batches.dead_ids.push(item.record.id.as_uuid());
                batches
                    .dead_tokens
                    .push(item.record.claim_token.map(|t| t.as_uuid()));
                batches.dead_errors.push(item.error.clone());

                batches
                    .dead_reasons
                    .push(crate::records::encode_dead_reason(reason));
            }
            // `FailureOutcome` is `#[non_exhaustive]` from another crate; a variant this
            // build does not know how to apply is left untouched rather than guessed at —
            // it stays claimed until its lease expires and is republished, the same benign
            // outcome as any other unresolved row (ADR 0008).
            _ => tracing::error!(
                id = %item.record.id,
                "unrecognised FailureOutcome variant; row left as-is"
            ),
        }
    }

    batches
}

/// Runs `fail_retry_rows`/`fail_dead_rows` for `batches` in one [`Session::run`]. Shared by
/// `fail`'s single batch pass above.
async fn apply_fail_batches(
    session: &Session,
    batches: &FailBatches,
) -> Result<(Vec<uuid::Uuid>, Vec<uuid::Uuid>), PostgresOutboxError> {
    session
        .run(async |conn: &mut PgConnection| {
            let applied_retry = if batches.retry_ids.is_empty() {
                Vec::new()
            } else {
                outcomes_repo::fail_retry_rows(
                    &mut *conn,
                    outcomes_repo::FailRetryRowsParams {
                        ids: &batches.retry_ids,
                        tokens: &batches.retry_tokens,
                        errors: &batches.retry_errors,
                        delays_ms: &batches.retry_delays,
                    },
                )
                .await?
            };
            let applied_dead = if batches.dead_ids.is_empty() {
                Vec::new()
            } else {
                outcomes_repo::fail_dead_rows(
                    &mut *conn,
                    outcomes_repo::FailDeadRowsParams {
                        ids: &batches.dead_ids,
                        tokens: &batches.dead_tokens,
                        errors: &batches.dead_errors,
                        reasons: &batches.dead_reasons,
                    },
                )
                .await?
            };

            Ok((applied_retry, applied_dead))
        })
        .await
        .map_err(|e| session.map_err::<PostgresOutboxError>(e))
}

/// Shared by `purge`'s three retention windows above.
fn to_millis(duration: Duration) -> i64 {
    i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
}