reliar-store-postgres 0.9.0

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

/// [`purge_published_rows`]'s bind parameters.
pub(in crate::outbox) struct PurgePublishedRowsParams {
    pub(in crate::outbox) retention_ms: i64,

    pub(in crate::outbox) batch_size: i64,
}

/// `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.
pub(in crate::outbox) async fn purge_published_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgePublishedRowsParams,
) -> 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')"#,
        params.retention_ms,
        params.batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// [`purge_dead_retention_rows`]'s bind parameters.
pub(in crate::outbox) struct PurgeDeadRetentionRowsParams {
    pub(in crate::outbox) retention_ms: i64,

    pub(in crate::outbox) batch_size: i64,
}

/// `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.
pub(in crate::outbox) async fn purge_dead_retention_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgeDeadRetentionRowsParams,
) -> 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')"#,
        params.retention_ms,
        params.batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// [`purge_expired_sweep_rows`]'s bind parameters.
pub(in crate::outbox) struct PurgeExpiredSweepRowsParams<'a> {
    pub(in crate::outbox) batch_size: i64,

    pub(in crate::outbox) dead_reason: &'a str,
}

/// `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. Also clears `claim_token` alongside `locked_by` —
/// this ends the claim (ADR 0046 Amendment A), so a stale outcome write quoting the swept row's
/// now-dead claim must find no token to match, not the row's real one.
///
/// **The lease guard is `(locked_by IS NULL OR available_at <= now())` — "not currently
/// leased" (ADR 0050 §2.3).** Both disjuncts are load-bearing: `available_at <= now()` alone
/// would postpone a *backed-off* expired row (`fail`→retry left `locked_by = NULL` and
/// `available_at` in the future); `locked_by IS NULL` alone would never reach a row whose lease
/// simply lapsed (nothing clears `locked_by` on expiry, only the next claim overwrites it).
pub(in crate::outbox) async fn purge_expired_sweep_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgeExpiredSweepRowsParams<'_>,
) -> 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,
                  claim_token  = 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_by IS NULL OR available_at <= now())
                 LIMIT $1
            )
              AND published_at IS NULL AND dead_at IS NULL
              AND (locked_by IS NULL OR available_at <= now())"#,
        params.batch_size,
        params.dead_reason,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// One row of [`stats_row`]'s statement.
pub(in crate::outbox) struct StatsRow {
    pub(in crate::outbox) pending: i64,

    pub(in crate::outbox) dead: i64,

    pub(in crate::outbox) expired_pending: i64,

    pub(in crate::outbox) oldest_pending_available_at: Option<time::OffsetDateTime>,

    pub(in crate::outbox) as_of: time::OffsetDateTime,
}

/// `stats`'s query (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. No bind parameters (carve-out 1, layout Part II §8.1).
pub(in crate::outbox) 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 (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 (expires_at IS NULL OR expires_at > now())
                ORDER BY available_at, id LIMIT 1)   AS oldest_pending_available_at,
             now()                                                          AS "as_of!""#
    )
    .fetch_one(executor)
    .await
}