reliar-store-postgres 0.7.0

//! Inbox contract §5.2 I-N2–I-N9 (ADR 0043 §3, §9 — re-tagged `unit` → `pg`): `InboxStore::process`'s
//! branches against real Postgres. The raw `claim`/`complete`/`fail` atomicity, redelivery and
//! concurrency properties already live in `inbox_claim.rs`/`inbox_concurrency.rs`/`inbox_fail.rs`;
//! this file is specifically about `process`'s own branching — running the handler, calling
//! `complete` automatically, and never doing either on the four non-`Claimed` claim answers.

use crate::common;
use crate::common::inbox::{
    InsertBusinessRow, InsertThenFail, SelfCompletingHandler, business_row_count,
    create_business_table, message,
};

use std::time::Duration;

use reliar_core::{Classify, FailureKind, MessageId};
use reliar_inbox::{InboxClaim, InboxOutcome, InboxProcessError, InboxScope, InboxStore};
use reliar_store_postgres::{PostgresInboxSettings, PostgresInboxStore};

/// I-N2 — `process` on a fresh key runs the handler once, calls `complete`, and returns
/// `Processed(v)` with the handler's own value.
async fn process_on_fresh_key_runs_handler_once_and_completes() {
    let pool = common::fresh_db().await;

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

    let outcome = store
        .process(
            &mut tx,
            &scope,
            message(id),
            &InsertBusinessRow { value: 1 },
        )
        .await
        .unwrap();

    assert_eq!(outcome, InboxOutcome::Processed(1));
    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());
}

/// I-N3 — `process` on a completed key never runs the handler, performs no write, and returns
/// `AlreadyCompleted`.
async fn process_on_completed_key_never_runs_handler() {
    let pool = common::fresh_db().await;

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

    let mut tx = pool.begin().await.unwrap();
    store
        .process(
            &mut tx,
            &scope,
            message(id),
            &InsertBusinessRow { value: 1 },
        )
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let record_before = store.find(&scope, id).await.unwrap().unwrap();

    let mut tx = pool.begin().await.unwrap();
    let outcome = store
        .process(
            &mut tx,
            &scope,
            message(id),
            &InsertBusinessRow { value: 2 },
        )
        .await
        .unwrap();

    assert!(matches!(outcome, InboxOutcome::AlreadyCompleted { .. }));
    tx.rollback().await.unwrap();

    assert_eq!(
        business_row_count(&pool).await,
        1,
        "the redelivery's handler must never run"
    );

    let record_after = store.find(&scope, id).await.unwrap().unwrap();
    assert_eq!(record_before.completed_at, record_after.completed_at);
    assert_eq!(record_before.attempts, record_after.attempts);
}

/// I-N4 — `process` against a key another (real) transaction is already claiming never runs the
/// handler, returns `InProgress` within a bounded wait (never a hang), and the caller's own
/// transaction is still usable afterwards.
async fn process_with_in_progress_claim_never_runs_handler_and_tx_stays_usable() {
    let pool = common::fresh_db().await;

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

    // tx_a holds the claim open — the real stimulus for InProgress, never a fabricated one.
    let mut tx_a = pool.begin().await.unwrap();
    store.claim(&mut tx_a, &scope, message(id)).await.unwrap();

    let mut tx_b = pool.begin().await.unwrap();
    let outcome = tokio::time::timeout(
        Duration::from_millis(500),
        store.process(
            &mut tx_b,
            &scope,
            message(id),
            &InsertBusinessRow { value: 99 },
        ),
    )
    .await
    .expect("process must return within 500ms instead of blocking on the advisory lock")
    .unwrap();

    assert_eq!(outcome, InboxOutcome::InProgress);

    // tx_b never ran the handler, and it is still usable — a second claim on a different key
    // succeeds through it.
    let other_id = MessageId::new();
    let claim = store
        .claim(&mut tx_b, &scope, message(other_id))
        .await
        .unwrap();
    assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });
    tx_b.rollback().await.unwrap();

    tx_a.rollback().await.unwrap();
    assert_eq!(
        business_row_count(&pool).await,
        0,
        "InProgress must never run InsertBusinessRow's handler"
    );
}

/// I-N5 — `process` when the handler errors returns `Err(Handler(e))`, does **not** call
/// `complete`, and preserves the source chain. Read inside the still-open transaction: a separate
/// connection would not see the uncommitted claim row at all.
async fn process_when_handler_errors_does_not_complete() {
    let pool = common::fresh_db().await;

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

    let err = store
        .process(&mut tx, &scope, message(id), &InsertThenFail { value: 1 })
        .await
        .unwrap_err();

    assert!(matches!(err, InboxProcessError::Handler(_)));
    assert!(
        std::error::Error::source(&err).is_some(),
        "the handler's error is the source"
    );

    let completed_at: Option<time::OffsetDateTime> =
        sqlx::query_scalar("SELECT completed_at FROM inbox WHERE scope = $1 AND message_id = $2")
            .bind(scope.as_str())
            .bind(id.as_uuid())
            .fetch_one(&mut *tx)
            .await
            .unwrap();
    assert!(
        completed_at.is_none(),
        "complete must not run after a handler error"
    );

    tx.rollback().await.unwrap();
    assert_eq!(
        business_row_count(&pool).await,
        0,
        "the handler's own write rolls back with the claim"
    );
}

/// I-N6 — `process` when `claim`/`complete` errors returns `Err(Store(e))`; `Classify` forwards
/// the store's own kind, and a `Handler` error always classifies `Transient`.
async fn process_store_error_forwards_kind_handler_error_is_transient() {
    let pool = common::fresh_db().await;

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

    // `process`'s own `complete` finds the row already completed — a real "no claimed row" fault,
    // not a fabricated one — because the handler completed it first, on the same transaction.
    let id = MessageId::new();
    let mut tx = pool.begin().await.unwrap();
    let handler = SelfCompletingHandler {
        store: &store,
        scope: scope.clone(),
        id,
    };

    let process_err = store
        .process(&mut tx, &scope, message(id), &handler)
        .await
        .unwrap_err();

    match &process_err {
        InboxProcessError::Store(err) => assert_eq!(err.kind(), FailureKind::Permanent),
        other => panic!("expected InboxProcessError::Store, got {other:?}"),
    }

    assert_eq!(
        process_err.kind(),
        FailureKind::Permanent,
        "Classify must forward the store's own kind, not override it"
    );
    tx.rollback().await.unwrap();

    // A handler failure always classifies Transient, regardless of the store's own kind.
    let failing_id = MessageId::new();
    let mut tx = pool.begin().await.unwrap();
    let process_err = store
        .process(
            &mut tx,
            &scope,
            message(failing_id),
            &InsertThenFail { value: 1 },
        )
        .await
        .unwrap_err();
    assert_eq!(process_err.kind(), FailureKind::Transient);
    tx.rollback().await.unwrap();
}

/// I-N7 — dropping the transaction without commit discards the claim row `process` wrote, so a
/// re-claim starts again at `Claimed { attempt: 1 }`. A real dropped connection/transaction, not a
/// fabricated rollback: `sqlx::Transaction::drop` issues the rollback itself.
async fn rollback_discards_claim_row_so_reclaim_restarts_at_attempt_one() {
    let pool = common::fresh_db().await;

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

    {
        let mut tx = pool.begin().await.unwrap();

        let outcome = store
            .process(
                &mut tx,
                &scope,
                message(id),
                &InsertBusinessRow { value: 1 },
            )
            .await
            .unwrap();

        assert_eq!(outcome, InboxOutcome::Processed(1));
        // `tx` drops here without `commit()` — a genuine rollback, not a stimulus.
    }

    assert!(
        store.find(&scope, id).await.unwrap().is_none(),
        "the claim row must not survive an uncommitted rollback"
    );
    assert_eq!(business_row_count(&pool).await, 0);

    let mut tx = pool.begin().await.unwrap();
    let outcome = store
        .process(
            &mut tx,
            &scope,
            message(id),
            &InsertBusinessRow { value: 1 },
        )
        .await
        .unwrap();
    assert_eq!(outcome, InboxOutcome::Processed(1));
    tx.commit().await.unwrap();

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

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "inbox_process::process_on_fresh_key_runs_handler_once_and_completes",
            move || {
                rt.block_on(process_on_fresh_key_runs_handler_once_and_completes());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_process::process_on_completed_key_never_runs_handler",
            move || {
                rt.block_on(process_on_completed_key_never_runs_handler());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_process::process_with_in_progress_claim_never_runs_handler_and_tx_stays_usable",
            move || {
                rt.block_on(
                    process_with_in_progress_claim_never_runs_handler_and_tx_stays_usable(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_process::process_when_handler_errors_does_not_complete",
            move || {
                rt.block_on(process_when_handler_errors_does_not_complete());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_process::process_store_error_forwards_kind_handler_error_is_transient",
            move || {
                rt.block_on(process_store_error_forwards_kind_handler_error_is_transient());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_process::rollback_discards_claim_row_so_reclaim_restarts_at_attempt_one",
            move || {
                rt.block_on(rollback_discards_claim_row_so_reclaim_restarts_at_attempt_one());
                Ok(())
            },
        ),
    ]
}