reliar-store-postgres 0.9.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; 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). ADR 0046 Amendment A.2's fencing `EXPLAIN` (conventions §6):
//! `complete`'s and `extend_lease`'s new `FROM UNNEST(...) AS f(id, token) WHERE o.id = f.id AND
//! o.claim_token = f.token` join must still plan as a nested loop over an index scan on
//! `pk_outbox`, never a sequential scan — the fencing predicate is a filter on a row the planner
//! already located by id, not a new access path. T7 (ADR 0050 §6): after `locked_until` drops out
//! of `ix_outbox_claimable`'s `INCLUDE` list, the claim plan is unchanged and `stats`' `pending`/
//! `oldest_pending_available_at` subqueries are still `Index Only Scan`s with zero heap fetches.

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, claim_token, \
                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}"
    );
}

/// P-29 (ADR 0049 Amendment A) — the claim, transcribed with `ORDER BY available_at, id`, still
/// plans on `ix_outbox_claimable` with no `Seq Scan` and — the new assertion this amendment adds —
/// no `Sort` node: `id` is a *key* column of the index, so the scan is already ordered, and a
/// `Sort` appearing would mean the index no longer matches the `ORDER BY`.
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 (expires_at IS NULL OR expires_at > now()) \
          ORDER BY available_at, id \
          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}"
    );
    assert!(
        !plan.contains("Seq Scan"),
        "the claim must never fall back to a sequential scan; plan:\n{plan}"
    );
    assert!(
        !plan.contains("Sort"),
        "id is a key column of ix_outbox_claimable, so the scan must already be ordered, with no \
         separate Sort node; plan:\n{plan}"
    );
}

/// A dedicated, genuinely mixed 100k-row seed for
/// [`stats_pending_and_oldest_pending_are_index_only_scans_with_no_heap_fetches`] — 20% each of
/// pending-due, leased, published, dead and expired-pending, mirroring the proportions
/// `PostgresOutboxStore::stats`'s own rustdoc measured (ADR 0040 §3). `shared_seeded_pool` above
/// is deliberately ~99% pending (the right shape for `list_dead`/the claim scan/purge), which
/// makes `pending`'s own `count(*)` cheaper as a `Seq Scan` than an `Index Only Scan` — a real,
/// cost-based planner choice this file's other trials don't contradict, but the wrong dataset to
/// prove the `INCLUDE`-list-shrink claim against.
async fn mixed_seeded_pool() -> sqlx::PgPool {
    let pool = common::fresh_db().await;
    let per_bucket: u32 = 20_000;

    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(per_bucket))
    .execute(&pool)
    .await
    .unwrap();

    sqlx::query(
        "INSERT INTO outbox (message_id, message_type, message_version, conversation_id, \
                              content_type, payload, available_at, created_at, locked_by) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now() + interval '1 hour', now(), 'perf-worker' \
           FROM generate_series(1, $1::bigint) AS g",
    )
    .bind(i64::from(per_bucket))
    .execute(&pool)
    .await
    .unwrap();

    sqlx::query(
        "INSERT INTO outbox (message_id, message_type, message_version, conversation_id, \
                              content_type, payload, available_at, created_at, published_at) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now(), now(), now() \
           FROM generate_series(1, $1::bigint) AS g",
    )
    .bind(i64::from(per_bucket))
    .execute(&pool)
    .await
    .unwrap();

    sqlx::query(
        "INSERT INTO outbox (message_id, message_type, message_version, conversation_id, \
                              content_type, payload, available_at, created_at, dead_at, \
                              dead_reason) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now(), now(), now(), 'permanent_error' \
           FROM generate_series(1, $1::bigint) AS g",
    )
    .bind(i64::from(per_bucket))
    .execute(&pool)
    .await
    .unwrap();

    sqlx::query(
        "INSERT INTO outbox (message_id, message_type, message_version, conversation_id, \
                              content_type, payload, available_at, created_at, expires_at) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now(), now(), now() - interval '1 second' \
           FROM generate_series(1, $1::bigint) AS g",
    )
    .bind(i64::from(per_bucket))
    .execute(&pool)
    .await
    .unwrap();

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

    pool
}

/// T7 (ADR 0050 §2.2, §6) — `stats`'s `pending` and `oldest_pending_available_at` subqueries are
/// still `Index Only Scan`s with zero heap fetches after the `INCLUDE` list shrank from
/// `(locked_until, expires_at)` to `(expires_at)` — dropping a column from `INCLUDE` narrows the
/// index, never widens the set of predicates it can answer without a heap fetch.
async fn stats_pending_and_oldest_pending_are_index_only_scans_with_no_heap_fetches() {
    let pool = mixed_seeded_pool().await;

    let pending_plan = explain(
        &pool,
        "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())",
    )
    .await;

    assert!(
        pending_plan.contains("Index Only Scan"),
        "expected stats' pending subquery to be an Index Only Scan; plan:\n{pending_plan}"
    );
    assert!(
        pending_plan.contains("Heap Fetches: 0"),
        "expected zero heap fetches on a vacuumed table; plan:\n{pending_plan}"
    );

    let oldest_plan = explain(
        &pool,
        "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",
    )
    .await;

    assert!(
        oldest_plan.contains("Index Only Scan"),
        "expected stats' oldest_pending_available_at subquery to be an Index Only Scan; \
         plan:\n{oldest_plan}"
    );
    assert!(
        oldest_plan.contains("Heap Fetches: 0"),
        "expected zero heap fetches on a vacuumed table; plan:\n{oldest_plan}"
    );
}

/// ADR 0046 Amendment A.2 — `complete`'s new `FROM UNNEST($1::uuid[], $2::uuid[]) AS f(id,
/// token) WHERE o.id = f.id AND o.claim_token = f.token` must still plan as a nested loop over an
/// index scan on `pk_outbox`, never a sequential scan of `outbox` — the fencing join adds a
/// filter on an already-located row, it must not change how that row is located.
async fn complete_plans_on_pk_outbox_with_no_seq_scan() {
    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();
    let tokens: Vec<Option<uuid::Uuid>> = ids.iter().map(|_| None).collect();

    // 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.
    // The bound tokens are all `NULL` — none of these rows is claimed — so every row is fenced
    // and the `UPDATE` touches nothing; the plan shape is identical whether the guard matches.
    let mut tx = pool.begin().await.unwrap();
    let plan: Vec<String> = sqlx::query_scalar(
        "EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) \
         UPDATE outbox o \
            SET published_at = now(), attempts = o.attempts + 1, locked_by = NULL, \
                claim_token = NULL, updated_at = now() \
           FROM UNNEST($1::uuid[], $2::uuid[]) AS f(id, token) \
          WHERE o.id = f.id AND o.claim_token = f.token",
    )
    .bind(&ids)
    .bind(&tokens)
    .fetch_all(&mut *tx)
    .await
    .unwrap();

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

    let plan = plan.join("\n");
    assert!(
        plan.contains("pk_outbox"),
        "expected complete's fenced UNNEST join to plan on pk_outbox; plan:\n{plan}"
    );
    assert!(
        plan.contains("Nested Loop"),
        "expected the UNNEST join to plan as a nested loop; plan:\n{plan}"
    );
    assert!(
        !plan.contains("Seq Scan on outbox"),
        "the fencing join must never fall back to a sequential scan of outbox; plan:\n{plan}"
    );
}

/// As [`complete_plans_on_pk_outbox_with_no_seq_scan`], for `extend_lease` — the same `UNNEST`
/// join shape, minus the terminal-write columns.
async fn extend_lease_plans_on_pk_outbox_with_no_seq_scan() {
    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();
    let tokens: Vec<Option<uuid::Uuid>> = ids.iter().map(|_| None).collect();

    let mut tx = pool.begin().await.unwrap();
    let plan: Vec<String> = sqlx::query_scalar(
        "EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) \
         UPDATE outbox o \
            SET available_at = now() + interval '30 seconds', updated_at = now() \
           FROM UNNEST($1::uuid[], $2::uuid[]) AS f(id, token) \
          WHERE o.id = f.id AND o.claim_token = f.token",
    )
    .bind(&ids)
    .bind(&tokens)
    .fetch_all(&mut *tx)
    .await
    .unwrap();

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

    let plan = plan.join("\n");
    assert!(
        plan.contains("pk_outbox"),
        "expected extend_lease's fenced UNNEST join to plan on pk_outbox; plan:\n{plan}"
    );
    assert!(
        plan.contains("Nested Loop"),
        "expected the UNNEST join to plan as a nested loop; plan:\n{plan}"
    );
    assert!(
        !plan.contains("Seq Scan on outbox"),
        "the fencing join must never fall back to a sequential scan of 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::stats_pending_and_oldest_pending_are_index_only_scans_with_no_heap_fetches",
            move || {
                rt.block_on(
                    stats_pending_and_oldest_pending_are_index_only_scans_with_no_heap_fetches(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_plans::complete_plans_on_pk_outbox_with_no_seq_scan",
            move || {
                rt.block_on(complete_plans_on_pk_outbox_with_no_seq_scan());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_plans::extend_lease_plans_on_pk_outbox_with_no_seq_scan",
            move || {
                rt.block_on(extend_lease_plans_on_pk_outbox_with_no_seq_scan());
                Ok(())
            },
        ),
    ]
}