reliar-store-postgres 0.9.0

//! Inbox contract §5.2 I-P1, I-P2, I-P3, I-P5 — same-transaction atomicity, the redelivery
//! (`AlreadyCompleted`) hot path performs no write, a crash between the handler and the commit
//! re-claims at `attempt: 1`, and two scopes on the same message id are independent. I-P33 (ADR
//! 0042 Amendment C.8) — `claim`'s step-3 fallback upsert cannot return zero rows, either branch.

use crate::common;
use crate::common::inbox::{InsertBusinessRow, business_row_count, create_business_table};
use crate::common::wait_until;

use std::time::Duration;

use reliar_core::MessageId;
use reliar_inbox::{InboxClaim, InboxHandler, InboxScope, InboxStore};
use reliar_store_postgres::{PostgresInboxSettings, PostgresInboxStore};
use sqlx::PgPool;

/// The row's `xmin` system column — its creating/last-updating transaction id. An `UPDATE`
/// always writes a new row version (a new `xmin`) even when every column's *value* is unchanged,
/// so an unchanged `xmin` across a claim is direct proof the claim issued no write at all — a
/// stronger, more precise check than only comparing `completed_at`, which an `UPDATE SET
/// completed_at = completed_at` would also leave unchanged.
async fn row_xmin(pool: &PgPool, scope: &str, message_id: uuid::Uuid) -> i64 {
    sqlx::query_scalar!(
        r#"SELECT xmin::text::bigint AS "xmin!" FROM inbox WHERE scope = $1 AND message_id = $2"#,
        scope,
        message_id,
    )
    .fetch_one(pool)
    .await
    .unwrap()
}

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

    create_business_table(&pool).await;
    let store =
        PostgresInboxStore::with_settings(pool.clone(), PostgresInboxSettings::default()).unwrap();
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();

    // Commit path: the business row and the inbox row commit together.
    let mut tx = pool.begin().await.unwrap();
    let claim = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

    let output = InsertBusinessRow { value: 1 }
        .handle(&mut tx)
        .await
        .unwrap();
    assert_eq!(output, 1);

    store.complete(&mut tx, &scope, id).await.unwrap();
    tx.commit().await.unwrap();

    assert_eq!(business_row_count(&pool).await, 1);
    let record = store.find(&scope, id).await.unwrap().unwrap();
    assert!(record.completed_at.is_some());

    // Rollback path, a different id: neither the business row nor the inbox row survives.
    let id2 = MessageId::new();
    let mut tx = pool.begin().await.unwrap();
    let claim = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id2))
        .await
        .unwrap();
    assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

    InsertBusinessRow { value: 2 }
        .handle(&mut tx)
        .await
        .unwrap();
    tx.rollback().await.unwrap();

    assert_eq!(
        business_row_count(&pool).await,
        1,
        "the rolled-back business row must not persist"
    );
    assert!(
        store.find(&scope, id2).await.unwrap().is_none(),
        "the rolled-back claim row must not persist"
    );
}

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

    create_business_table(&pool).await;
    let store =
        PostgresInboxStore::with_settings(pool.clone(), PostgresInboxSettings::default()).unwrap();
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();

    let mut tx = pool.begin().await.unwrap();
    store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    InsertBusinessRow { value: 1 }
        .handle(&mut tx)
        .await
        .unwrap();
    store.complete(&mut tx, &scope, id).await.unwrap();
    tx.commit().await.unwrap();

    let first = store.find(&scope, id).await.unwrap().unwrap();
    let xmin_before = row_xmin(&pool, scope.as_str(), id.as_uuid()).await;

    // A second delivery: claim reports AlreadyCompleted, the caller rolls back and never re-runs
    // the handler — the business table stays at exactly one row.
    let mut tx = pool.begin().await.unwrap();
    let claim = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();

    match claim {
        InboxClaim::AlreadyCompleted { completed_at } => {
            assert_eq!(completed_at, first.completed_at.unwrap());
        }
        other => panic!("expected AlreadyCompleted, got {other:?}"),
    }

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

    assert_eq!(business_row_count(&pool).await, 1);

    let second = store.find(&scope, id).await.unwrap().unwrap();
    assert_eq!(
        second.completed_at, first.completed_at,
        "the redelivery's claim performed no write"
    );
    assert_eq!(
        row_xmin(&pool, scope.as_str(), id.as_uuid()).await,
        xmin_before,
        "an unchanged xmin proves the redelivery's claim issued no UPDATE at all, not merely one \
         that happened to write the same completed_at back"
    );

    // A third claim after the redelivery still reports the same AlreadyCompleted answer.
    let mut tx = pool.begin().await.unwrap();
    assert!(matches!(
        store
            .claim(&mut tx, &scope, crate::common::inbox::message(id))
            .await
            .unwrap(),
        InboxClaim::AlreadyCompleted { .. }
    ));
    tx.rollback().await.unwrap();
}

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

    create_business_table(&pool).await;
    let store =
        PostgresInboxStore::with_settings(pool.clone(), PostgresInboxSettings::default()).unwrap();
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();

    {
        let mut tx = pool.begin().await.unwrap();
        let claim = store
            .claim(&mut tx, &scope, crate::common::inbox::message(id))
            .await
            .unwrap();

        assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

        InsertBusinessRow { value: 1 }
            .handle(&mut tx)
            .await
            .unwrap();

        // Simulated crash: `tx` is dropped without a commit or an explicit rollback. `sqlx`'s
        // `Transaction::drop` issues the rollback itself.
    }

    assert_eq!(
        business_row_count(&pool).await,
        0,
        "the dropped transaction's business write must not persist"
    );
    assert!(
        store.find(&scope, id).await.unwrap().is_none(),
        "the dropped transaction's claim row must not persist"
    );

    // Redelivery: the row went with the rollback, so this looks like a first delivery.
    let mut tx = pool.begin().await.unwrap();
    let claim = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

    let output = InsertBusinessRow { value: 1 }
        .handle(&mut tx)
        .await
        .unwrap();
    assert_eq!(output, 1);
    store.complete(&mut tx, &scope, id).await.unwrap();
    tx.commit().await.unwrap();

    assert_eq!(business_row_count(&pool).await, 1);
}

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

    create_business_table(&pool).await;
    let store =
        PostgresInboxStore::with_settings(pool.clone(), PostgresInboxSettings::default()).unwrap();
    let id = MessageId::new();
    let scope_a = InboxScope::new("projection-a").unwrap();
    let scope_b = InboxScope::new("projection-b").unwrap();

    for (scope, value) in [(&scope_a, 1), (&scope_b, 2)] {
        let mut tx = pool.begin().await.unwrap();
        let claim = store
            .claim(&mut tx, scope, crate::common::inbox::message(id))
            .await
            .unwrap();

        assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

        InsertBusinessRow { value }.handle(&mut tx).await.unwrap();
        store.complete(&mut tx, scope, id).await.unwrap();
        tx.commit().await.unwrap();
    }

    assert_eq!(business_row_count(&pool).await, 2);
    assert!(
        store
            .find(&scope_a, id)
            .await
            .unwrap()
            .unwrap()
            .completed_at
            .is_some()
    );
    assert!(
        store
            .find(&scope_b, id)
            .await
            .unwrap()
            .unwrap()
            .completed_at
            .is_some()
    );
}

/// The "commit without `complete`" duplicate window (inbox contract §semantics, dated 2026-09-07):
/// a caller that commits after `claim` alone — never calling `complete` — leaves a committed,
/// uncompleted row indistinguishable from one `fail` created. The next redelivery's `claim`
/// answers `Claimed { attempt: 1 }` again and the handler re-runs.
async fn commit_without_complete_redelivers_at_attempt_one() {
    let pool = common::fresh_db().await;

    create_business_table(&pool).await;
    let store =
        PostgresInboxStore::with_settings(pool.clone(), PostgresInboxSettings::default()).unwrap();
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();

    let mut tx = pool.begin().await.unwrap();
    let claim = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

    InsertBusinessRow { value: 1 }
        .handle(&mut tx)
        .await
        .unwrap();
    // Commits without ever calling `complete` — the misuse this window documents.
    tx.commit().await.unwrap();

    let record = store.find(&scope, id).await.unwrap().unwrap();
    assert!(
        record.completed_at.is_none(),
        "the committed row is uncompleted, exactly like one `fail` would have created"
    );

    let mut tx = pool.begin().await.unwrap();
    let redelivery = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    assert_eq!(
        redelivery,
        InboxClaim::Claimed { attempt: 1 },
        "the redelivery's claim cannot tell this row apart from a fresh one"
    );
    InsertBusinessRow { value: 2 }
        .handle(&mut tx)
        .await
        .unwrap();
    store.complete(&mut tx, &scope, id).await.unwrap();
    tx.commit().await.unwrap();

    assert_eq!(
        business_row_count(&pool).await,
        2,
        "the handler re-ran and inserted its business row a second time"
    );
}

/// True once some *other* backend on this database is waiting on a lock — mirrors
/// `inbox_purge.rs`'s helper of the same name.
async fn some_other_backend_is_waiting_on_a_lock(pool: &PgPool) -> bool {
    let count: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM pg_stat_activity \
          WHERE wait_event_type = 'Lock' AND pid <> pg_backend_pid() \
            AND datname = current_database()",
    )
    .fetch_one(pool)
    .await
    .unwrap();

    count > 0
}

/// I-P33 (ADR 0042 Amendment C.8) — `claim`'s step-3 fallback statement is
/// `INSERT … ON CONFLICT (scope, message_id) DO UPDATE SET updated_at = inbox.updated_at
/// RETURNING id, attempts, completed_at, dead_at`: PostgreSQL's documented atomic
/// insert-or-update guarantee for a `DO UPDATE` with no `WHERE` clause means this can never
/// return zero rows. Proven directly against the raw statement (not through `claim`, which
/// cannot be forced into step 3's fallback from the public API — there is no wait point between
/// step 2 and step 3 for a `purge` to land in), against a row a third connection holds open with
/// `SELECT … FOR UPDATE`, exactly as `inbox_purge.rs`'s own race fixture does:
///
/// (a) holder commits a `DELETE` of that row ⇒ the statement blocks on the holder, then
///     **inserts** and returns exactly one row for the id this call minted.
/// (b) holder releases without deleting ⇒ the statement blocks, then **updates** and returns
///     exactly one row carrying the row's **existing** id and `attempts` — never the id this
///     call minted, which is the trap the rustdoc calls out — and `updated_at` **unchanged**: the
///     `SET updated_at = inbox.updated_at` is an identity assignment (C.8), so the update branch
///     must not move the column the incomplete sweep ages by (I-P31).
async fn upsert_statement<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    id: uuid::Uuid,
    scope: &str,
    message_id: uuid::Uuid,
) -> (
    uuid::Uuid,
    i32,
    Option<time::OffsetDateTime>,
    Option<time::OffsetDateTime>,
    time::OffsetDateTime,
) {
    let row = sqlx::query!(
        r#"INSERT INTO inbox (id, scope, message_id, message_type, message_version,
                               conversation_id, correlation_id, causation_id)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
           ON CONFLICT (scope, message_id) DO UPDATE
              SET updated_at = inbox.updated_at
           RETURNING id, attempts, completed_at, dead_at, updated_at"#,
        id,
        scope,
        message_id,
        "orders.created",
        1_i32,
        message_id,
        None::<&str>,
        None::<uuid::Uuid>,
    )
    .fetch_one(executor)
    .await
    .unwrap();

    (
        row.id,
        row.attempts,
        row.completed_at,
        row.dead_at,
        row.updated_at,
    )
}

#[allow(
    clippy::too_many_lines,
    reason = "one race fixture proving both branches of a single atomic statement — splitting it \
              would scatter one ordered narrative (seed, block, release, assert) across helper \
              functions with no reuse"
)]
async fn step_3_upsert_never_returns_zero_rows_either_branch() {
    let pool = common::fresh_db().await;
    let scope = "orders-projection";

    // (a) the holder's row is deleted while blocked: the upsert must insert.
    {
        let existing_id = uuid::Uuid::now_v7();
        let message_id = uuid::Uuid::now_v7();

        sqlx::query!(
            "INSERT INTO inbox (id, scope, message_id, message_type, message_version, \
                                 conversation_id, attempts) \
             VALUES ($1, $2, $3, 'orders.created', 1, $3, 5)",
            existing_id,
            scope,
            message_id,
        )
        .execute(&pool)
        .await
        .unwrap();

        let mut holder = pool.begin().await.unwrap();

        sqlx::query!(
            r#"SELECT message_id FROM inbox WHERE scope = $1 AND message_id = $2 FOR UPDATE"#,
            scope,
            message_id,
        )
        .fetch_all(&mut *holder)
        .await
        .unwrap();

        let new_id = uuid::Uuid::now_v7();
        let upsert_task = tokio::spawn({
            let pool = pool.clone();

            async move { upsert_statement(&pool, new_id, scope, message_id).await }
        });

        wait_until(
            "another backend is waiting on the row lock",
            Duration::from_secs(5),
            || some_other_backend_is_waiting_on_a_lock(&pool),
        )
        .await;

        sqlx::query!("DELETE FROM inbox WHERE id = $1", existing_id)
            .execute(&mut *holder)
            .await
            .unwrap();
        holder.commit().await.unwrap();

        let (returned_id, attempts, completed_at, dead_at, _updated_at) =
            upsert_task.await.expect("upsert task did not panic");

        assert_eq!(
            returned_id, new_id,
            "the delete-then-insert branch returns the id this call minted"
        );
        assert_eq!(attempts, 0, "a fresh insert starts at 0 attempts");
        assert!(completed_at.is_none());
        assert!(dead_at.is_none());
    }

    // (b) the holder releases without deleting: the upsert must update, returning the
    // **existing** row's id and attempts, never the id this call minted, and `updated_at`
    // unchanged (I-P33's identity-`SET` assertion).
    {
        let existing_id = uuid::Uuid::now_v7();
        let message_id = uuid::Uuid::now_v7();

        let seeded_updated_at = sqlx::query_scalar!(
            "INSERT INTO inbox (id, scope, message_id, message_type, message_version, \
                                 conversation_id, attempts) \
             VALUES ($1, $2, $3, 'orders.created', 1, $3, 5) \
             RETURNING updated_at",
            existing_id,
            scope,
            message_id,
        )
        .fetch_one(&pool)
        .await
        .unwrap();

        let mut holder = pool.begin().await.unwrap();

        sqlx::query!(
            r#"SELECT message_id FROM inbox WHERE scope = $1 AND message_id = $2 FOR UPDATE"#,
            scope,
            message_id,
        )
        .fetch_all(&mut *holder)
        .await
        .unwrap();

        let new_id = uuid::Uuid::now_v7();
        let upsert_task = tokio::spawn({
            let pool = pool.clone();

            async move { upsert_statement(&pool, new_id, scope, message_id).await }
        });

        wait_until(
            "another backend is waiting on the row lock",
            Duration::from_secs(5),
            || some_other_backend_is_waiting_on_a_lock(&pool),
        )
        .await;

        holder.commit().await.unwrap();

        let (returned_id, attempts, completed_at, dead_at, updated_at) =
            upsert_task.await.expect("upsert task did not panic");

        assert_eq!(
            returned_id, existing_id,
            "the update branch returns the existing row's id, never the id this call minted"
        );
        assert_eq!(
            attempts, 5,
            "the identity SET leaves attempts untouched by the update branch"
        );
        assert!(completed_at.is_none());
        assert!(dead_at.is_none());
        assert_eq!(
            updated_at, seeded_updated_at,
            "the identity SET updated_at = inbox.updated_at must not move the column (C.8)"
        );
    }
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "inbox_claim::same_transaction_atomicity_commit_and_rollback",
            move || {
                rt.block_on(same_transaction_atomicity_commit_and_rollback());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_claim::redelivery_after_commit_is_already_completed_and_performs_no_write",
            move || {
                rt.block_on(redelivery_after_commit_is_already_completed_and_performs_no_write());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_claim::crash_between_handler_and_commit_reclaims_at_attempt_one",
            move || {
                rt.block_on(crash_between_handler_and_commit_reclaims_at_attempt_one());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_claim::two_scopes_on_one_message_are_independent",
            move || {
                rt.block_on(two_scopes_on_one_message_are_independent());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_claim::commit_without_complete_redelivers_at_attempt_one",
            move || {
                rt.block_on(commit_without_complete_redelivers_at_attempt_one());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_claim::step_3_upsert_never_returns_zero_rows_either_branch",
            move || {
                rt.block_on(step_3_upsert_never_returns_zero_rows_either_branch());
                Ok(())
            },
        ),
    ]
}