reliar_store_postgres/inbox/
inbox_store_dead_letters.rs

1//! [`reliar_inbox::InboxDeadLetters`] for [`PostgresInboxStore`] (inbox contract §3.1, ADR 0042
2//! A.2.5). Three statements, all on the provider's own pool; `Tx` is inert for every one of them.
3//! The whole store-layer policy lives in this trait impl, reading `self.session` directly and
4//! calling `dead_letters`'s query functions for their statements.
5
6use sqlx::PgConnection;
7
8use reliar_inbox::{InboxDeadLetters, InboxDeadQuery, InboxRecord, InboxRecordId};
9use tracing::Instrument as _;
10
11use super::dead_letters as repo;
12use super::error::PostgresInboxError;
13use super::inbox_store::build_record;
14
15use super::PostgresInboxStore;
16
17/// The largest [`InboxDeadQuery::limit`] [`InboxDeadLetters::list_dead`] honours — a
18/// caller-supplied value above this is silently capped, never sent to the database, mirroring
19/// [`crate::PostgresOutboxStore`]'s own `MAX_LIST_DEAD_LIMIT` (outbox contract; inbox contract
20/// §5.2 I-P23).
21const MAX_LIST_DEAD_LIMIT: u32 = 1000;
22
23impl InboxDeadLetters for PostgresInboxStore {
24    type Error = PostgresInboxError;
25
26    /// `ORDER BY dead_at, id` is normative: database-authored death time is the keyset's leading
27    /// column, and the unique row id breaks ties.
28    // Block form — reason (a): the `reliar.inbox.list_dead` span's entry fields must be recorded
29    // before the statement runs.
30    fn list_dead(
31        &self,
32        query: InboxDeadQuery,
33    ) -> impl Future<Output = Result<Vec<InboxRecord>, Self::Error>> + Send {
34        // Provider-capped at `MAX_LIST_DEAD_LIMIT` (1000) — distinct from `InboxDeadQuery`'s own
35        // `limit` default of 100: a caller-supplied `limit` above the cap never reaches the
36        // database, whatever value `InboxDeadQuery` carries.
37        let capped_limit = query.limit.min(MAX_LIST_DEAD_LIMIT);
38        let limit = i64::from(capped_limit);
39        let scope_owned = query.scope.clone();
40        let dead_before = query.dead_before;
41        let (after_dead_at, after_id) = query.after.map_or((None, None), |cursor| {
42            (Some(cursor.dead_at()), Some(cursor.id().as_uuid()))
43        });
44        let message_type = query.message_type.clone();
45
46        let span = tracing::debug_span!(
47            "reliar.inbox.list_dead",
48            inbox.scope = scope_owned.as_ref().map(reliar_inbox::InboxScope::as_str),
49            inbox.limit = capped_limit,
50            inbox.returned = tracing::field::Empty,
51        );
52        let recording_span = span.clone();
53
54        async move {
55            let scope = scope_owned.as_ref().map(reliar_inbox::InboxScope::as_str);
56
57            let rows = self
58                .session
59                .run(async |conn: &mut PgConnection| {
60                    repo::list_dead_rows(
61                        &mut *conn,
62                        repo::ListDeadRowsParams {
63                            scope,
64                            message_type: message_type.as_deref(),
65                            dead_before,
66                            after_dead_at,
67                            after_id,
68                            limit,
69                        },
70                    )
71                    .await
72                })
73                .await
74                .map_err(|e| self.session.map_err::<PostgresInboxError>(e))?;
75
76            recording_span.record("inbox.returned", rows.len());
77
78            rows.into_iter().map(build_record).collect()
79        }
80        .instrument(span)
81    }
82
83    // Block form — reason (a): the `reliar.inbox.retry_dead` span's `inbox.requested` field must
84    // be recorded before the statement runs.
85    fn retry_dead(
86        &self,
87        ids: &[InboxRecordId],
88    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
89        let requested = ids.len();
90        let ids: Vec<uuid::Uuid> = ids
91            .iter()
92            .map(reliar_inbox::InboxRecordId::as_uuid)
93            .collect();
94
95        let span = tracing::debug_span!(
96            "reliar.inbox.retry_dead",
97            inbox.requested = requested,
98            inbox.affected = tracing::field::Empty,
99        );
100        let recording_span = span.clone();
101
102        async move {
103            if ids.is_empty() {
104                recording_span.record("inbox.affected", 0_u64);
105
106                return Ok(0);
107            }
108
109            let result = self
110                .session
111                .run(async |conn: &mut PgConnection| {
112                    repo::retry_dead_rows(&mut *conn, repo::RetryDeadRowsParams { ids: &ids }).await
113                })
114                .await
115                .map_err(|e| self.session.map_err::<PostgresInboxError>(e))?;
116
117            recording_span.record("inbox.affected", result);
118
119            Ok(result)
120        }
121        .instrument(span)
122    }
123
124    // Block form — reason (a): the `reliar.inbox.purge_dead` span's `inbox.requested` field must
125    // be recorded before the statement runs.
126    fn purge_dead(
127        &self,
128        ids: &[InboxRecordId],
129    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
130        let requested = ids.len();
131        let ids: Vec<uuid::Uuid> = ids
132            .iter()
133            .map(reliar_inbox::InboxRecordId::as_uuid)
134            .collect();
135
136        let span = tracing::debug_span!(
137            "reliar.inbox.purge_dead",
138            inbox.requested = requested,
139            inbox.affected = tracing::field::Empty,
140        );
141        let recording_span = span.clone();
142
143        async move {
144            if ids.is_empty() {
145                recording_span.record("inbox.affected", 0_u64);
146
147                return Ok(0);
148            }
149
150            let result = self
151                .session
152                .run(async |conn: &mut PgConnection| {
153                    repo::purge_dead_rows(&mut *conn, repo::PurgeDeadRowsParams { ids: &ids }).await
154                })
155                .await
156                .map_err(|e| self.session.map_err::<PostgresInboxError>(e))?;
157
158            recording_span.record("inbox.affected", result);
159
160            Ok(result)
161        }
162        .instrument(span)
163    }
164}
reliar_store_postgres/inbox/inbox_store_dead_letters.rs

reliar_store_postgres/inbox/
inbox_store_dead_letters.rs
