reliar-store-postgres 0.7.0

PostgreSQL provider for the Reliar transactional outbox and inbox: migrations, migrate(), enqueue and the SKIP LOCKED claim.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

//! `PostgresInboxSettings::statement_timeout` bounding `fail`'s pool-side
//! `INSERT … ON CONFLICT DO UPDATE`: with a concurrent claimer's row still uncommitted, `fail`'s
//! speculative insert would otherwise block for that claimer's entire handler duration (the exact
//! wait `claim`'s advisory-lock probe exists to avoid on the claim side, reintroduced on the
//! recovery path — see `src/inbox/outcomes.rs::fail`'s doc). A non-zero timeout turns that
//! unbounded wait into a clean, typed `FailureKind::Transient` error instead.

use crate::common;
use crate::common::inbox::HandlerFailed;

use std::time::Duration;

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

async fn fail_blocks_behind_an_uncommitted_claim_until_the_timeout_cancels_it_then_succeeds() {
    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();

    // Connection A: claims the key but never commits — `fail`'s `INSERT … ON CONFLICT DO UPDATE`
    // has no choice but to wait on this still-open row.
    let mut holder = pool.begin().await.unwrap();
    let claim = store
        .claim(&mut holder, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();

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

    let settings = PostgresInboxSettings::default().statement_timeout(Duration::from_millis(200));
    let timeout_store = PostgresInboxStore::connect(pool.clone(), settings)
        .await
        .unwrap();

    // The outer bound proves this is a *clean, timely* error, not a hang the test would otherwise
    // wait on indefinitely; the `statement_timeout` itself is what actually resolves it, well
    // inside this bound.
    let result = tokio::time::timeout(
        Duration::from_secs(5),
        timeout_store.fail(&scope, crate::common::inbox::message(id), &HandlerFailed),
    )
    .await
    .expect("must return well within the outer bound, not hang");

    let err = result.expect_err("a canceled statement must surface as a typed error");
    assert!(
        matches!(err, PostgresInboxError::Database { .. }),
        "expected Database, got {err:?}"
    );
    assert_eq!(
        err.kind(),
        FailureKind::Transient,
        "a canceled statement is safe to retry"
    );

    // Once A commits, the row is no longer contended and `fail` succeeds normally.
    holder.commit().await.unwrap();
    timeout_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,
        "the successful fail() recorded its attempt"
    );
}

async fn a_generous_non_zero_timeout_behaves_identically_to_zero_for_fail_find_and_purge() {
    let pool = common::fresh_db().await;
    let settings = PostgresInboxSettings::default().statement_timeout(Duration::from_secs(30));
    let store = PostgresInboxStore::connect(pool.clone(), settings)
        .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, "fail behaves the same wrapped");

    let report = store
        .purge(
            reliar_inbox::InboxPurgeRequest::default().incomplete_retention(Some(Duration::ZERO)),
        )
        .await
        .unwrap();
    assert_eq!(
        report.incomplete_deleted, 1,
        "purge behaves the same wrapped"
    );
}

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