reliar-store-postgres 0.6.0

//! `purge`/`stats`: the bounded retention sweep and the four-subquery stats snapshot.

use reliar_core::Serializer;
use reliar_outbox::{OutboxStats, PurgeReport, PurgeRequest};

use crate::error::PostgresStoreError;

use super::PostgresOutboxStore;

/// [`reliar_outbox::OutboxStore::purge`]'s body.
pub(super) async fn purge<Ser: Serializer + Send + Sync + 'static>(
    store: &PostgresOutboxStore<Ser>,
    request: PurgeRequest,
) -> Result<PurgeReport, PostgresStoreError> {
    let batch_size = i64::from(request.batch_size);
    let expired_reason = crate::records::encode_dead_reason(reliar_outbox::DeadReason::Expired);

    let (published_deleted, dead_deleted, expired_to_dead) =
        if store.settings.statement_timeout.is_zero() {
            let published_deleted = if let Some(retention) = request.published_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_published_rows(&store.pool, retention_ms, batch_size)
                    .await
                    .map_err(|e| store.map_err(e))?
            } else {
                0
            };

            let dead_deleted = if let Some(retention) = request.dead_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_dead_retention_rows(&store.pool, retention_ms, batch_size)
                    .await
                    .map_err(|e| store.map_err(e))?
            } else {
                0
            };

            let expired_to_dead = purge_expired_sweep_rows(&store.pool, batch_size, expired_reason)
                .await
                .map_err(|e| store.map_err(e))?;

            (published_deleted, dead_deleted, expired_to_dead)
        } else {
            // One transaction, one `SET LOCAL statement_timeout`, all three statements —
            // each is individually bounded by it (`statement_timeout` bounds every statement
            // Reliar issues on its own pool, `purge` included), and sharing one transaction
            // costs one `BEGIN`/`SET LOCAL`/`COMMIT` round trip instead of three.
            let mut tx = store.pool.begin().await.map_err(|e| store.map_err(e))?;

            store.set_local_timeout(&mut tx).await?;

            let published_deleted = if let Some(retention) = request.published_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_published_rows(&mut *tx, retention_ms, batch_size)
                    .await
                    .map_err(|e| store.map_err(e))?
            } else {
                0
            };

            let dead_deleted = if let Some(retention) = request.dead_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_dead_retention_rows(&mut *tx, retention_ms, batch_size)
                    .await
                    .map_err(|e| store.map_err(e))?
            } else {
                0
            };

            let expired_to_dead = purge_expired_sweep_rows(&mut *tx, batch_size, expired_reason)
                .await
                .map_err(|e| store.map_err(e))?;

            tx.commit().await.map_err(|e| store.map_err(e))?;

            (published_deleted, dead_deleted, expired_to_dead)
        };

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

/// [`reliar_outbox::OutboxStore::stats`]'s body.
pub(super) async fn stats<Ser: Serializer + Send + Sync + 'static>(
    store: &PostgresOutboxStore<Ser>,
) -> Result<OutboxStats, PostgresStoreError> {
    if store.settings.statement_timeout.is_zero() {
        let row = stats_row(&store.pool).await.map_err(|e| store.map_err(e))?;

        return 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,
        ));
    }

    // Same one statement, wrapped in a `SET LOCAL statement_timeout` transaction.
    let mut tx = store.pool.begin().await.map_err(|e| store.map_err(e))?;
    store.set_local_timeout(&mut tx).await?;
    let row = stats_row(&mut *tx).await.map_err(|e| store.map_err(e))?;
    tx.commit().await.map_err(|e| store.map_err(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,
    ))
}

/// `purge`'s published-row delete, bounded by `batch_size`. The outer `WHERE` repeats the
/// subselect's own predicate **in full** — not just the `IS NOT NULL` half — so `EvalPlanQual`'s
/// re-check (on a row a concurrent writer touched between the subselect's snapshot and this
/// statement's lock acquisition) can actually exclude it, rather than deleting on stale
/// information — the retention-age comparison needs repeating too, not only nullness, since a
/// row could in principle be re-published with a fresher timestamp between the snapshot and the
/// lock.
async fn purge_published_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    retention_ms: i64,
    batch_size: i64,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"DELETE FROM outbox WHERE id IN (
               SELECT id FROM outbox
                WHERE published_at IS NOT NULL
                  AND published_at < now() - ($1::bigint * interval '1 millisecond')
                LIMIT $2
           )
           AND published_at IS NOT NULL
           AND published_at < now() - ($1::bigint * interval '1 millisecond')"#,
        retention_ms,
        batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// `purge`'s dead-row delete, bounded by `batch_size`. Same full-predicate `EvalPlanQual` guard
/// as [`purge_published_rows`] — without it, a row `retry_dead` resurrects (or re-deadens with a
/// fresher `dead_at`) between the subselect's snapshot and this statement's lock acquisition
/// could still be deleted; the same repeated guard also covers the retention-age comparison,
/// not only nullness.
async fn purge_dead_retention_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    retention_ms: i64,
    batch_size: i64,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"DELETE FROM outbox WHERE id IN (
               SELECT id FROM outbox
                WHERE dead_at IS NOT NULL
                  AND dead_at < now() - ($1::bigint * interval '1 millisecond')
                LIMIT $2
           )
           AND dead_at IS NOT NULL
           AND dead_at < now() - ($1::bigint * interval '1 millisecond')"#,
        retention_ms,
        batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// `purge`'s expired-pending-to-dead sweep, bounded by `batch_size`. The outer
/// `published_at IS NULL AND dead_at IS NULL` plus the lease clause repeat the subselect's own
/// mutable-state predicates so a lapsed-lease worker's concurrent `complete`/`fail` can't race
/// this into a `ck_outbox_terminal` violation; `expires_at` itself is immutable
/// once written, so it doesn't need repeating.
async fn purge_expired_sweep_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    batch_size: i64,
    expired_reason: &str,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"UPDATE outbox
              SET dead_at      = now(),
                  dead_reason  = $2,
                  last_error   = 'reliar: expired before publication',
                  locked_by    = NULL,
                  locked_until = NULL,
                  updated_at   = now()
            WHERE id IN (
                SELECT id FROM outbox
                 WHERE expires_at IS NOT NULL AND expires_at < now()
                   AND published_at IS NULL AND dead_at IS NULL
                   AND (locked_until IS NULL OR locked_until < now())
                 LIMIT $1
            )
              AND published_at IS NULL AND dead_at IS NULL
              AND (locked_until IS NULL OR locked_until < now())"#,
        batch_size,
        expired_reason,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// `stats`'s query, shared by the plain-pool and `statement_timeout`-wrapped-transaction call
/// sites (ADR 0040 §3): four independently planned scalar subqueries in one statement, each aimed
/// at its own partial index rather than one `Seq Scan` computing every aggregate together.
async fn stats_row<'e>(executor: impl sqlx::PgExecutor<'e>) -> Result<StatsRow, sqlx::Error> {
    sqlx::query_as!(
        StatsRow,
        r#"SELECT
             (SELECT count(*) FROM outbox
                WHERE published_at IS NULL AND dead_at IS NULL
                  AND available_at <= now()
                  AND (locked_until IS NULL OR locked_until < now())
                  AND (expires_at IS NULL OR expires_at > now()))            AS "pending!",
             (SELECT count(*) FROM outbox WHERE dead_at IS NOT NULL)         AS "dead!",
             (SELECT count(*) FROM outbox
                WHERE published_at IS NULL AND dead_at IS NULL
                  AND expires_at IS NOT NULL AND expires_at < now())         AS "expired_pending!",
             (SELECT available_at FROM outbox
                WHERE published_at IS NULL AND dead_at IS NULL
                  AND available_at <= now()
                  AND (locked_until IS NULL OR locked_until < now())
                  AND (expires_at IS NULL OR expires_at > now())
                ORDER BY available_at, sequence LIMIT 1)   AS oldest_pending_available_at,
             now()                                                          AS "as_of!""#
    )
    .fetch_one(executor)
    .await
}

/// One row of [`stats_row`]'s statement.
struct StatsRow {
    pending: i64,

    dead: i64,

    expired_pending: i64,

    oldest_pending_available_at: Option<time::OffsetDateTime>,

    as_of: time::OffsetDateTime,
}