reliar-store-postgres 0.6.0

//! N6/N7/N9 — business data and an outbox row written in one `sqlx` transaction are both present
//! after commit and both absent after rollback; a reused `MessageId` aborts the transaction and
//! discards an earlier business write in it; the future is `Send` through `tokio::spawn` against a
//! real `Transaction<'_, Postgres>` (decision #37/#38, ADR 0036 amendment B). Replaces
//! `outbox_enqueue_atomic.rs` and `outbox_publisher_enqueue.rs` — after RELIAR-61 there is no
//! facade type and `OutboxEnqueue` has one method, so the two files' scenarios collapsed into one.

use crate::common;

use crate::common::OrderCreated;
use reliar_core::Envelope;
use reliar_core::Serializer as _;
use reliar_outbox::OutboxEnqueue;
use reliar_store_postgres::PostgresOutboxStore;

async fn count_business_rows(pool: &sqlx::PgPool) -> i64 {
    sqlx::query_scalar("SELECT count(*) FROM widgets")
        .fetch_one(pool)
        .await
        .unwrap()
}

async fn count_outbox_rows(pool: &sqlx::PgPool) -> i64 {
    sqlx::query_scalar("SELECT count(*) FROM outbox")
        .fetch_one(pool)
        .await
        .unwrap()
}

async fn create_widgets_table(pool: &sqlx::PgPool) {
    sqlx::query("CREATE TABLE widgets (id bigserial PRIMARY KEY)")
        .execute(pool)
        .await
        .unwrap();
}

/// N6 — a business row and an outbox row written in one transaction are both visible after
/// commit, and the committed row carries the store's serializer's `content_type` and its
/// serialized bytes.
async fn commit_makes_both_rows_visible() {
    let pool = common::fresh_db().await;

    create_widgets_table(&pool).await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();

    let mut tx = pool.begin().await.unwrap();
    sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await
        .unwrap();
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    let envelope_id = envelope.id;
    let expected_payload = reliar_core::JsonSerializer
        .serialize(&envelope.body)
        .unwrap();
    store.enqueue_envelope(&mut tx, envelope).await.unwrap();
    tx.commit().await.unwrap();

    assert_eq!(count_business_rows(&pool).await, 1);
    assert_eq!(count_outbox_rows(&pool).await, 1);

    let content_type: String = sqlx::query_scalar("SELECT content_type FROM outbox WHERE id = $1")
        .bind(envelope_id.as_uuid())
        .fetch_one(&pool)
        .await
        .unwrap();
    let payload: Vec<u8> = sqlx::query_scalar("SELECT payload FROM outbox WHERE id = $1")
        .bind(envelope_id.as_uuid())
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(content_type, store.content_type().to_string());
    assert_eq!(payload, expected_payload.as_ref());
}

/// N6 — rollback leaves neither row visible.
async fn rollback_makes_neither_row_visible() {
    let pool = common::fresh_db().await;

    create_widgets_table(&pool).await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();

    let mut tx = pool.begin().await.unwrap();
    sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await
        .unwrap();
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    store.enqueue_envelope(&mut tx, envelope).await.unwrap();
    tx.rollback().await.unwrap();

    assert_eq!(count_business_rows(&pool).await, 0);
    assert_eq!(count_outbox_rows(&pool).await, 0);
}

/// N7 — a reused `MessageId` maps to [`reliar_store_postgres::EnqueueError::Duplicate`], leaves
/// the transaction aborted (the next statement on it fails too), and an earlier business write
/// made in that same transaction is gone once the caller gives up on it — "an earlier `Ok` is not
/// durable"; "treat any enqueue error as *abort this transaction*".
async fn duplicate_message_id_aborts_the_transaction_and_discards_the_earlier_business_write() {
    let pool = common::fresh_db().await;

    create_widgets_table(&pool).await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    let envelope_id = envelope.id;

    // Committed ahead of time, in its own transaction — its id is what the second `enqueue` below
    // reuses, forcing that `INSERT` (and therefore the whole transaction) to abort. The envelope
    // is cloned explicitly for the reuse — by-value `enqueue` never clones on the caller's behalf.
    let mut seed_tx = pool.begin().await.unwrap();
    store
        .enqueue_envelope(&mut seed_tx, envelope.clone())
        .await
        .unwrap();
    seed_tx.commit().await.unwrap();

    let mut tx = pool.begin().await.unwrap();
    sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await
        .unwrap();

    let result = store.enqueue_envelope(&mut tx, envelope).await;

    match result.expect_err("a reused MessageId must be rejected") {
        reliar_store_postgres::EnqueueError::Duplicate { id } => {
            assert_eq!(id, envelope_id);
        }
        other => panic!("expected EnqueueError::Duplicate, got {other:?}"),
    }

    // The transaction is already aborted server-side: the next statement on it fails too.
    let next_statement = sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await;
    assert!(
        next_statement.is_err(),
        "a statement on an aborted transaction must fail"
    );

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

    // The earlier business write never survives — the whole transaction was all-or-nothing.
    assert_eq!(
        count_business_rows(&pool).await,
        0,
        "an earlier write in the aborted transaction must not be durable"
    );
}

/// N9 — the R23 regression guard: the future `enqueue` returns is `Send` even though it borrows a
/// non-`'static` `Transaction<'_, Postgres>` scope — `tokio::spawn` requires exactly that, and
/// this is the one place a change to `T: Message + Sync` on the typed method (decision #37) would
/// show up as a compile failure, not just a runtime one.
async fn enqueue_is_send_through_tokio_spawn() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();

    let envelope = Envelope::builder(OrderCreated { order_id: 7 }).build();
    let envelope_id = envelope.id;
    let mut tx = pool.begin().await.unwrap();

    tokio::spawn(async move {
        store.enqueue_envelope(&mut tx, envelope).await.unwrap();

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

    let count: i64 = sqlx::query_scalar("SELECT count(*) FROM outbox WHERE id = $1")
        .bind(envelope_id.as_uuid())
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 1);
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "outbox_enqueue::commit_makes_both_rows_visible",
            move || {
                rt.block_on(commit_makes_both_rows_visible());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue::rollback_makes_neither_row_visible",
            move || {
                rt.block_on(rollback_makes_neither_row_visible());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue::duplicate_message_id_aborts_the_transaction_and_discards_the_earlier_business_write",
            move || {
                rt.block_on(
                    duplicate_message_id_aborts_the_transaction_and_discards_the_earlier_business_write(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue::enqueue_is_send_through_tokio_spawn",
            move || {
                rt.block_on(enqueue_is_send_through_tokio_spawn());
                Ok(())
            },
        ),
    ]
}