reliar-store-postgres 0.9.0

//! Inbox contract §5.2 I-P14/I-P27/I-P29, §3.3 — the expected plan shapes, measured after
//! `VACUUM (ANALYZE)`: `claim`'s state read, `complete` and `find` plan as an index scan on
//! `ix_inbox_scope_message_id` (ADR 0042 A.2.1 moved the primary key to a surrogate `id`), never
//! a `Seq Scan`; the completed-row purge sub-select plans against `ix_inbox_completed`; the
//! dead-row purge sub-select plans against `ix_inbox_dead`; `list_dead`'s keyset listing
//! (I-P29, ADR 0042 Amendment C.1) plans as an index scan on `ix_inbox_dead` with no `Sort`
//! node; the incomplete-row purge sub-select is an accepted `Seq Scan` (no index backs it,
//! and — Amendment C.2 — no `attempts > 0` predicate either) and only its `LIMIT` bound is
//! asserted.

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 `const`/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.
///
/// `EXPLAIN ANALYZE` actually **executes** `statement` — for a read this is inert, but for the
/// `UPDATE` this file also measures it is a real write. Generic over `impl PgExecutor` (not
/// hardcoded to `&PgPool`) precisely so a caller measuring a write can pass a transaction it
/// rolls back afterward instead, leaving the seeded data untouched for whatever runs next.
async fn explain<'e>(executor: impl sqlx::PgExecutor<'e>, statement: &str) -> String {
    let lines: Vec<String> = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
        "EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) {statement}"
    )))
    .fetch_all(executor)
    .await
    .unwrap();

    lines.join("\n")
}

/// Seeds `n` rows via one `UNNEST`-driven bulk `INSERT` (a network round trip per row would make
/// a 50k-row seed the slow part of the suite), most of them recently completed — so a completed
/// purge's retention predicate stays **selective** and the planner has a reason to prefer the
/// partial index over scanning the whole table, exactly the shape a real backlog has. Mirrors
/// `outbox_claim_index_scale.rs`'s "seed a realistic mix, not a uniform one" approach.
async fn seed(pool: &sqlx::PgPool, n: u32) {
    let row_ids: Vec<uuid::Uuid> = (0..n).map(|_| uuid::Uuid::now_v7()).collect();
    let message_ids: Vec<uuid::Uuid> = (0..n).map(|_| uuid::Uuid::now_v7()).collect();

    sqlx::query(
        "INSERT INTO inbox (id, scope, message_id, message_type, message_version, \
                             conversation_id, completed_at) \
         SELECT id, 'orders-projection', message_id, 'orders.created', 1, message_id, \
                now() - interval '1 hour' \
           FROM UNNEST($1::uuid[], $2::uuid[]) AS t(id, message_id)",
    )
    .bind(&row_ids)
    .bind(&message_ids)
    .execute(pool)
    .await
    .unwrap();

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

/// Backdates a small, selective subset of `seed`'s rows past the retention window, so the
/// completed-purge sub-select's predicate matches a small fraction of the table rather than
/// (nearly) all of it.
async fn age_a_few_rows_past_retention(pool: &sqlx::PgPool, n: u32) {
    sqlx::query(
        "UPDATE inbox SET completed_at = now() - interval '10 days' \
          WHERE message_id IN (SELECT message_id FROM inbox ORDER BY message_id LIMIT $1)",
    )
    .bind(i64::from(n))
    .execute(pool)
    .await
    .unwrap();

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

/// Large enough that PostgreSQL's own cost model prefers an index over a full scan — the same
/// "100k+ rows" floor `outbox_claim_index_scale.rs` measured at; a few hundred rows fit in so
/// few pages that a `Seq Scan` is genuinely cheaper and correctly chosen, which is why this file
/// no longer seeds at that scale.
const SEED_ROWS: u32 = 100_000;

async fn ix_inbox_scope_message_id_backs_the_state_read_complete_and_find() {
    let pool = common::fresh_db().await;

    seed(&pool, SEED_ROWS).await;

    let target: uuid::Uuid = sqlx::query_scalar("SELECT message_id FROM inbox LIMIT 1")
        .fetch_one(&pool)
        .await
        .unwrap();

    // The `UPDATE` among these three is a real write once `EXPLAIN ANALYZE` runs it — a
    // transaction, rolled back at the end, keeps the seeded data untouched for whatever runs
    // next rather than actually completing the target row.
    let mut tx = pool.begin().await.unwrap();

    for statement in [
        format!(
            "SELECT id, attempts, completed_at, dead_at FROM inbox WHERE scope = 'orders-projection' AND message_id = '{target}'"
        ),
        format!(
            "UPDATE inbox SET completed_at = now(), updated_at = now() \
             WHERE scope = 'orders-projection' AND message_id = '{target}' AND completed_at IS NULL AND dead_at IS NULL"
        ),
        format!(
            "SELECT id, scope, message_id, message_type, message_version, conversation_id, \
                    correlation_id, causation_id, received_at, updated_at, completed_at, dead_at, \
                    attempts, last_error \
             FROM inbox WHERE scope = 'orders-projection' AND message_id = '{target}'"
        ),
    ] {
        let plan = explain(&mut *tx, &statement).await;

        assert!(
            plan.contains("ix_inbox_scope_message_id"),
            "expected ix_inbox_scope_message_id to back the lookup; plan:\n{plan}"
        );
        assert!(
            !plan.contains("Seq Scan"),
            "expected no Seq Scan for a two-column unique-index lookup; plan:\n{plan}"
        );
    }

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

async fn dead_purge_sub_select_uses_ix_inbox_dead() {
    let pool = common::fresh_db().await;

    seed(&pool, SEED_ROWS).await;
    sqlx::query(
        "UPDATE inbox SET dead_at = now() - interval '10 days', completed_at = NULL \
          WHERE message_id IN (SELECT message_id FROM inbox ORDER BY message_id LIMIT 1000)",
    )
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query("VACUUM (ANALYZE) inbox")
        .execute(&pool)
        .await
        .unwrap();

    let plan = explain(
        &pool,
        "SELECT id FROM inbox \
          WHERE dead_at IS NOT NULL AND dead_at < now() - interval '7 days' \
          LIMIT 1000",
    )
    .await;

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

/// I-P29 (ADR 0042 Amendment E) — `list_dead`'s `(dead_at, id)` keyset listing plans as an index
/// scan on `ix_inbox_dead` with **no `Sort` node** and no `Seq Scan on inbox`, `LIMIT`-bounded.
async fn list_dead_uses_ix_inbox_dead_with_no_sort() {
    let pool = common::fresh_db().await;

    seed(&pool, SEED_ROWS).await;
    sqlx::query(
        "UPDATE inbox SET dead_at = now() - interval '1 hour', completed_at = NULL \
          WHERE message_id IN (SELECT message_id FROM inbox ORDER BY message_id LIMIT 1000)",
    )
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query("VACUUM (ANALYZE) inbox")
        .execute(&pool)
        .await
        .unwrap();

    // Mirrors `list_dead_rows`'s shipped statement exactly (column list, all four optional
    // 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, scope, message_id, message_type, message_version, conversation_id, \
                correlation_id, causation_id, received_at, updated_at, completed_at, dead_at, \
                attempts, last_error \
           FROM inbox \
          WHERE dead_at IS NOT NULL \
            AND (NULL::text IS NULL OR scope = NULL) \
            AND (NULL::text IS NULL OR message_type = 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, id \
          LIMIT 100",
    )
    .await;

    assert!(
        plan.contains("ix_inbox_dead"),
        "expected ix_inbox_dead 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 inbox"),
        "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 completed_purge_sub_select_uses_ix_inbox_completed() {
    let pool = common::fresh_db().await;

    seed(&pool, SEED_ROWS).await;
    age_a_few_rows_past_retention(&pool, 1_000).await;

    let plan = explain(
        &pool,
        "SELECT id FROM inbox \
          WHERE completed_at IS NOT NULL \
            AND completed_at < now() - interval '7 days' \
          LIMIT 1000",
    )
    .await;

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

async fn incomplete_purge_sub_select_is_an_accepted_seq_scan_bounded_by_limit() {
    let pool = common::fresh_db().await;

    seed(&pool, 200).await;

    let plan = explain(
        &pool,
        "SELECT id FROM inbox \
          WHERE completed_at IS NULL AND dead_at IS NULL \
            AND updated_at < now() - interval '1 hour' \
          LIMIT 1000",
    )
    .await;

    assert!(
        plan.contains("Seq Scan"),
        "the incomplete purge sub-select has no supporting index by design; plan:\n{plan}"
    );
    assert!(
        plan.contains("Limit"),
        "the sub-select must still be bounded by LIMIT; plan:\n{plan}"
    );
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "inbox_plans::ix_inbox_scope_message_id_backs_the_state_read_complete_and_find",
            move || {
                rt.block_on(ix_inbox_scope_message_id_backs_the_state_read_complete_and_find());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_plans::completed_purge_sub_select_uses_ix_inbox_completed",
            move || {
                rt.block_on(completed_purge_sub_select_uses_ix_inbox_completed());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_plans::incomplete_purge_sub_select_is_an_accepted_seq_scan_bounded_by_limit",
            move || {
                rt.block_on(incomplete_purge_sub_select_is_an_accepted_seq_scan_bounded_by_limit());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_plans::dead_purge_sub_select_uses_ix_inbox_dead",
            move || {
                rt.block_on(dead_purge_sub_select_uses_ix_inbox_dead());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_plans::list_dead_uses_ix_inbox_dead_with_no_sort",
            move || {
                rt.block_on(list_dead_uses_ix_inbox_dead_with_no_sort());
                Ok(())
            },
        ),
    ]
}