reliar-store-postgres 0.7.0

//! ADR 0044 §7 P-12 — the expected plan shapes after the identity split, measured after
//! `VACUUM (ANALYZE)`: `list_dead`'s keyset listing plans as an index scan on
//! `ix_outbox_dead_cursor` with no `Sort` node and no `Seq Scan`; the dead-retention purge's
//! sub-select plans against the same index; `complete`'s `WHERE id = ANY(...)` plans on
//! `pk_outbox`; the claim still plans on `ix_outbox_claimable` (the existing scale trial in
//! `outbox_claim_index_scale.rs` must not regress — this is the same claim statement, mirrored).

use crate::common;

/// Runs `EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) <statement>` and returns the plan as one
/// string. `statement` is always one of this file's own literal-substituted SQL strings, never
/// caller/user input — the same sanctioned `AssertSqlSafe` exception `outbox_claim_index_scale.rs`
/// already uses for test-only dynamic SQL.
async fn explain(pool: &sqlx::PgPool, statement: &str) -> String {
    let lines: Vec<String> = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
        "EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) {statement}"
    )))
    .fetch_all(pool)
    .await
    .unwrap();

    lines.join("\n")
}

/// Large enough that PostgreSQL's own cost model prefers an index over a full scan — the same
/// floor `outbox_claim_index_scale.rs` and `inbox_plans.rs` both measure at.
const SEED_ROWS: u32 = 100_000;

/// Seeds `n` mostly-pending rows via one `UNNEST`-driven bulk `INSERT`, with a selective subset
/// aged into `dead_at IS NOT NULL` — so the dead-letter listing and purge sub-select both have a
/// real, selective predicate to plan against, the same "realistic mix, not a uniform one"
/// approach `outbox_claim_index_scale.rs`/`inbox_plans.rs` use.
async fn seed(pool: &sqlx::PgPool, n: u32) {
    sqlx::query(
        "INSERT INTO outbox (message_id, message_type, message_version, conversation_id, \
                              content_type, payload, available_at, created_at) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now() - (g || ' seconds')::interval, now() - (g || ' seconds')::interval \
           FROM generate_series(1, $1::bigint) AS g",
    )
    .bind(i64::from(n))
    .execute(pool)
    .await
    .unwrap();

    sqlx::query(
        "UPDATE outbox SET dead_at = now() - interval '1 hour', dead_reason = 'permanent_error' \
          WHERE id IN (SELECT id FROM outbox ORDER BY id LIMIT 1000)",
    )
    .execute(pool)
    .await
    .unwrap();

    sqlx::query("VACUUM (ANALYZE) outbox")
        .execute(pool)
        .await
        .unwrap();
}

/// The 100k-row seed is the expensive part of every trial below and none of them mutates it
/// (`complete_plans_on_pk_outbox` runs its `UPDATE` inside a transaction it rolls back) — built
/// once and shared, review round 1 nit, mirroring `common::db`'s own migrated-template-per-test
/// pattern one level up.
static SHARED_SEEDED_POOL: tokio::sync::OnceCell<sqlx::PgPool> = tokio::sync::OnceCell::const_new();

async fn shared_seeded_pool() -> sqlx::PgPool {
    SHARED_SEEDED_POOL
        .get_or_init(|| async {
            let pool = common::fresh_db().await;

            seed(&pool, SEED_ROWS).await;

            pool
        })
        .await
        .clone()
}

async fn list_dead_uses_ix_outbox_dead_cursor_with_no_sort() {
    let pool = shared_seeded_pool().await;

    // Mirrors `list_dead_rows`'s shipped statement exactly (column list, all filters as their
    // `IS NULL OR …` shape, composite ordering, `LIMIT $6`) — an unfiltered call is the shape
    // every filter's `IS NULL` branch takes, not a stripped-down stand-in for it.
    let plan = explain(
        &pool,
        "SELECT id, message_id, message_type, message_version, \
                correlation_id, conversation_id, causation_id, request_id, \
                content_type, payload, tenant_id, expires_at, ordering_key, \
                metadata, headers, metadata_version, \
                created_at, available_at, \
                attempts, locked_by, locked_until, \
                published_at, dead_at, dead_reason, last_error \
           FROM outbox \
          WHERE dead_at IS NOT NULL \
            AND (NULL::text IS NULL OR message_type = NULL) \
            AND (NULL::text IS NULL OR tenant_id = NULL) \
            AND (NULL::timestamptz IS NULL OR dead_at < NULL) \
            AND (NULL::timestamptz IS NULL OR (dead_at, id) > (NULL, NULL::uuid)) \
          ORDER BY dead_at ASC, id ASC \
          LIMIT 100",
    )
    .await;

    assert!(
        plan.contains("ix_outbox_dead_cursor"),
        "expected ix_outbox_dead_cursor to back list_dead's keyset listing; plan:\n{plan}"
    );
    assert!(
        !plan.contains("Sort"),
        "list_dead must read in death-time order directly off the index, never via a Sort node; \
         plan:\n{plan}"
    );
    assert!(
        !plan.contains("Seq Scan on outbox"),
        "list_dead must never fall back to a sequential scan; plan:\n{plan}"
    );
    assert!(
        plan.contains("Limit"),
        "list_dead must still be bounded by LIMIT; plan:\n{plan}"
    );
}

async fn dead_retention_purge_sub_select_uses_ix_outbox_dead_cursor() {
    let pool = shared_seeded_pool().await;

    // Mirrors `purge_dead_retention_rows`'s sub-select shape.
    let plan = explain(
        &pool,
        "SELECT id FROM outbox \
          WHERE dead_at IS NOT NULL \
            AND dead_at < now() - interval '30 days' \
          LIMIT 1000",
    )
    .await;

    assert!(
        plan.contains("ix_outbox_dead_cursor"),
        "expected ix_outbox_dead_cursor to back the dead-retention purge sub-select; plan:\n{plan}"
    );
}

async fn claim_still_plans_on_ix_outbox_claimable() {
    let pool = shared_seeded_pool().await;

    // Hand-transcribed from `claim_rows`'s CTE `SELECT` (bind params replaced with literals so
    // `EXPLAIN` can run with no arguments) — the identity split touches only the `RETURNING`
    // list, never the claim scan's own predicate or ordering.
    let plan = explain(
        &pool,
        "SELECT id 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 50 \
          FOR UPDATE SKIP LOCKED",
    )
    .await;

    assert!(
        plan.contains("ix_outbox_claimable"),
        "expected ix_outbox_claimable to still back the claim scan; plan:\n{plan}"
    );
}

async fn complete_plans_on_pk_outbox() {
    let pool = shared_seeded_pool().await;

    let ids: Vec<uuid::Uuid> = sqlx::query_scalar("SELECT id FROM outbox LIMIT 10")
        .fetch_all(&pool)
        .await
        .unwrap();

    // Mirrors `complete_rows`'s shipped statement exactly, run inside a transaction that is
    // rolled back afterward so `EXPLAIN ANALYZE`'s real `UPDATE` leaves the seeded data untouched.
    let mut tx = pool.begin().await.unwrap();
    let plan: Vec<String> = sqlx::query_scalar(
        "EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) \
         UPDATE outbox \
            SET published_at = now(), attempts = attempts + 1, locked_by = NULL, \
                locked_until = NULL, updated_at = now() \
          WHERE id = ANY($1) AND locked_by = $2",
    )
    .bind(&ids)
    .bind("no-such-worker")
    .fetch_all(&mut *tx)
    .await
    .unwrap();

    tx.rollback().await.unwrap();

    let plan = plan.join("\n");
    assert!(
        plan.contains("pk_outbox"),
        "expected complete's WHERE id = ANY(...) to plan on pk_outbox; plan:\n{plan}"
    );
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "outbox_plans::list_dead_uses_ix_outbox_dead_cursor_with_no_sort",
            move || {
                rt.block_on(list_dead_uses_ix_outbox_dead_cursor_with_no_sort());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_plans::dead_retention_purge_sub_select_uses_ix_outbox_dead_cursor",
            move || {
                rt.block_on(dead_retention_purge_sub_select_uses_ix_outbox_dead_cursor());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_plans::claim_still_plans_on_ix_outbox_claimable",
            move || {
                rt.block_on(claim_still_plans_on_ix_outbox_claimable());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test("outbox_plans::complete_plans_on_pk_outbox", move || {
            rt.block_on(complete_plans_on_pk_outbox());
            Ok(())
        }),
    ]
}