reliar-store-postgres 0.7.0

//! Inbox contract §5.2 I-P6, I-P7, I-P8 — `fail` bookkeeping (increments `attempts`, keeps the
//! latest error, feeds the next `claim`'s `attempt`), a stale `fail` on an already-completed row
//! is a no-op, and `complete` without a preceding claim is `NotClaimed` (permanent). I-N8 (ADR
//! 0043 §3, folded from `reliar-inbox`'s removed fake-backed `fail.rs`) — `last_error`'s 2 KiB
//! truncation with its marker, matching `InboxRecord::last_error`'s documented bound.

use crate::common;

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

#[derive(Debug)]
struct HandlerFailed;

impl std::fmt::Display for HandlerFailed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "handler failed, by design")
    }
}

impl std::error::Error for HandlerFailed {}

#[derive(Debug)]
struct LongError(String);

impl std::fmt::Display for LongError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for LongError {}

async fn fail_increments_attempts_and_keeps_latest_error() {
    let pool = common::fresh_db().await;
    let store = PostgresInboxStore::connect(pool.clone(), PostgresInboxSettings::default())
        .await
        .unwrap();
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();

    store
        .fail(&scope, crate::common::inbox::message(id), &HandlerFailed)
        .await
        .unwrap();
    let record = store.find(&scope, id).await.unwrap().unwrap();
    assert_eq!(record.attempts, 1);
    assert!(record.completed_at.is_none());
    assert_eq!(
        record.last_error.as_deref(),
        Some("handler failed, by design")
    );

    store
        .fail(&scope, crate::common::inbox::message(id), &HandlerFailed)
        .await
        .unwrap();
    let record = store.find(&scope, id).await.unwrap().unwrap();
    assert_eq!(
        record.attempts, 2,
        "a second fail() must count as a second attempt"
    );

    // The next redelivery claims at the recorded attempts + 1.
    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: 3 });
    tx.rollback().await.unwrap();
}

async fn fail_on_a_completed_row_is_a_no_op() {
    let pool = common::fresh_db().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
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    store.complete(&mut tx, &scope, id).await.unwrap();
    tx.commit().await.unwrap();

    let before = store.find(&scope, id).await.unwrap().unwrap();
    assert!(before.completed_at.is_some());
    assert_eq!(before.attempts, 0);
    assert!(before.last_error.is_none());

    store
        .fail(&scope, crate::common::inbox::message(id), &HandlerFailed)
        .await
        .unwrap();

    let after = store.find(&scope, id).await.unwrap().unwrap();
    assert_eq!(
        after.completed_at, before.completed_at,
        "a stale fail() must not touch completed_at"
    );
    assert_eq!(after.attempts, before.attempts, "or attempts");
    assert_eq!(after.last_error, before.last_error, "or last_error");
}

/// I-N8's truncation half — `fail`'s truncation matches `InboxRecord::last_error`'s documented
/// 2 KiB bound.
async fn fail_truncates_a_long_error_at_2kib_with_a_marker() {
    let pool = common::fresh_db().await;
    let store = PostgresInboxStore::connect(pool.clone(), PostgresInboxSettings::default())
        .await
        .unwrap();
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();
    let long = LongError("x".repeat(4096));

    store
        .fail(&scope, crate::common::inbox::message(id), &long)
        .await
        .unwrap();

    let record = store.find(&scope, id).await.unwrap().unwrap();
    let last_error = record.last_error.unwrap();
    assert!(last_error.len() <= 2048);
    assert!(last_error.ends_with("…[truncated]"));
}

async fn complete_without_a_claim_is_not_claimed_and_permanent() {
    let pool = common::fresh_db().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.complete(&mut tx, &scope, id).await.unwrap_err();

    assert_eq!(err.kind(), FailureKind::Permanent);
    tx.rollback().await.unwrap();
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "inbox_fail::fail_increments_attempts_and_keeps_latest_error",
            move || {
                rt.block_on(fail_increments_attempts_and_keeps_latest_error());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_fail::fail_on_a_completed_row_is_a_no_op",
            move || {
                rt.block_on(fail_on_a_completed_row_is_a_no_op());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_fail::fail_truncates_a_long_error_at_2kib_with_a_marker",
            move || {
                rt.block_on(fail_truncates_a_long_error_at_2kib_with_a_marker());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_fail::complete_without_a_claim_is_not_claimed_and_permanent",
            move || {
                rt.block_on(complete_without_a_claim_is_not_claimed_and_permanent());
                Ok(())
            },
        ),
    ]
}