reliar_store_postgres/inbox/
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
4use reliar_inbox::{InboxDeadLetters, InboxDeadQuery, InboxRecord, InboxRecordId};
5use tracing::Instrument as _;
6
7use super::error::PostgresInboxError;
8
9use super::PostgresInboxStore;
10use super::purge::{InboxRow, build_record};
11
12/// The largest [`InboxDeadQuery::limit`] [`InboxDeadLetters::list_dead`] honours — a
13/// caller-supplied value above this is silently capped, never sent to the database, mirroring
14/// [`crate::PostgresOutboxStore`]'s own `MAX_LIST_DEAD_LIMIT` (outbox contract; inbox contract
15/// §5.2 I-P23).
16const MAX_LIST_DEAD_LIMIT: u32 = 1000;
17
18impl InboxDeadLetters for PostgresInboxStore {
19    type Error = PostgresInboxError;
20
21    /// `ORDER BY dead_at, id` is normative: database-authored death time is the keyset's leading
22    /// column, and the unique row id breaks ties.
23    // Block form — reason (a): the `reliar.inbox.list_dead` span's entry fields must be recorded
24    // before the statement runs.
25    fn list_dead(
26        &self,
27        query: InboxDeadQuery,
28    ) -> impl Future<Output = Result<Vec<InboxRecord>, Self::Error>> + Send {
29        // Provider-capped at `MAX_LIST_DEAD_LIMIT` (1000) — distinct from `InboxDeadQuery`'s own
30        // `limit` default of 100: a caller-supplied `limit` above the cap never reaches the
31        // database, whatever value `InboxDeadQuery` carries.
32        let capped_limit = query.limit.min(MAX_LIST_DEAD_LIMIT);
33        let limit = i64::from(capped_limit);
34        let scope_owned = query.scope.clone();
35        let dead_before = query.dead_before;
36        let (after_dead_at, after_id) = query.after.map_or((None, None), |cursor| {
37            (Some(cursor.dead_at()), Some(cursor.id().as_uuid()))
38        });
39        let message_type = query.message_type.clone();
40
41        let span = tracing::debug_span!(
42            "reliar.inbox.list_dead",
43            inbox.scope = scope_owned.as_ref().map(reliar_inbox::InboxScope::as_str),
44            inbox.limit = capped_limit,
45            inbox.returned = tracing::field::Empty,
46        );
47        let recording_span = span.clone();
48
49        async move {
50            let scope = scope_owned.as_ref().map(reliar_inbox::InboxScope::as_str);
51
52            let rows = if self.settings.statement_timeout.is_zero() {
53                list_dead_rows(
54                    &self.pool,
55                    scope,
56                    message_type.as_deref(),
57                    dead_before,
58                    after_dead_at,
59                    after_id,
60                    limit,
61                )
62                .await?
63            } else {
64                let mut tx = self.pool.begin().await?;
65
66                self.set_local_timeout(&mut tx).await?;
67                let rows = list_dead_rows(
68                    &mut *tx,
69                    scope,
70                    message_type.as_deref(),
71                    dead_before,
72                    after_dead_at,
73                    after_id,
74                    limit,
75                )
76                .await?;
77                tx.commit().await?;
78
79                rows
80            };
81
82            recording_span.record("inbox.returned", rows.len());
83
84            rows.into_iter().map(build_record).collect()
85        }
86        .instrument(span)
87    }
88
89    // Block form — reason (a): the `reliar.inbox.retry_dead` span's `inbox.requested` field must
90    // be recorded before the statement runs.
91    fn retry_dead(
92        &self,
93        ids: &[InboxRecordId],
94    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
95        let requested = ids.len();
96        let ids: Vec<uuid::Uuid> = ids
97            .iter()
98            .map(reliar_inbox::InboxRecordId::as_uuid)
99            .collect();
100
101        let span = tracing::debug_span!(
102            "reliar.inbox.retry_dead",
103            inbox.requested = requested,
104            inbox.affected = tracing::field::Empty,
105        );
106        let recording_span = span.clone();
107
108        async move {
109            if ids.is_empty() {
110                recording_span.record("inbox.affected", 0_u64);
111
112                return Ok(0);
113            }
114
115            let result = if self.settings.statement_timeout.is_zero() {
116                retry_dead_rows(&self.pool, &ids).await?
117            } else {
118                let mut tx = self.pool.begin().await?;
119
120                self.set_local_timeout(&mut tx).await?;
121                let result = retry_dead_rows(&mut *tx, &ids).await?;
122                tx.commit().await?;
123
124                result
125            };
126
127            recording_span.record("inbox.affected", result);
128
129            Ok(result)
130        }
131        .instrument(span)
132    }
133
134    // Block form — reason (a): the `reliar.inbox.purge_dead` span's `inbox.requested` field must
135    // be recorded before the statement runs.
136    fn purge_dead(
137        &self,
138        ids: &[InboxRecordId],
139    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
140        let requested = ids.len();
141        let ids: Vec<uuid::Uuid> = ids
142            .iter()
143            .map(reliar_inbox::InboxRecordId::as_uuid)
144            .collect();
145
146        let span = tracing::debug_span!(
147            "reliar.inbox.purge_dead",
148            inbox.requested = requested,
149            inbox.affected = tracing::field::Empty,
150        );
151        let recording_span = span.clone();
152
153        async move {
154            if ids.is_empty() {
155                recording_span.record("inbox.affected", 0_u64);
156
157                return Ok(0);
158            }
159
160            let result = if self.settings.statement_timeout.is_zero() {
161                purge_dead_rows(&self.pool, &ids).await?
162            } else {
163                let mut tx = self.pool.begin().await?;
164
165                self.set_local_timeout(&mut tx).await?;
166                let result = purge_dead_rows(&mut *tx, &ids).await?;
167                tx.commit().await?;
168
169                result
170            };
171
172            recording_span.record("inbox.affected", result);
173
174            Ok(result)
175        }
176        .instrument(span)
177    }
178}
179
180#[allow(
181    clippy::too_many_arguments,
182    reason = "each argument is one InboxDeadQuery filter; a struct wrapper would just move the \
183              same six names one level down"
184)]
185async fn list_dead_rows<'e>(
186    executor: impl sqlx::PgExecutor<'e>,
187    scope: Option<&str>,
188    message_type: Option<&str>,
189    dead_before: Option<time::OffsetDateTime>,
190    after_dead_at: Option<time::OffsetDateTime>,
191    after_id: Option<uuid::Uuid>,
192    limit: i64,
193) -> Result<Vec<InboxRow>, sqlx::Error> {
194    sqlx::query_as!(
195        InboxRow,
196        r#"SELECT id, scope, message_id, message_type, message_version, conversation_id,
197                  correlation_id, causation_id, received_at, updated_at, completed_at, dead_at,
198                  attempts, last_error
199             FROM inbox
200            WHERE dead_at IS NOT NULL
201              AND ($1::text IS NULL OR scope = $1)
202              AND ($2::text IS NULL OR message_type = $2)
203              AND ($3::timestamptz IS NULL OR dead_at < $3)
204              AND ($4::timestamptz IS NULL OR (dead_at, id) > ($4, $5::uuid))
205            ORDER BY dead_at, id
206            LIMIT $6"#,
207        scope,
208        message_type,
209        dead_before,
210        after_dead_at,
211        after_id,
212        limit,
213    )
214    .fetch_all(executor)
215    .await
216}
217
218async fn retry_dead_rows<'e>(
219    executor: impl sqlx::PgExecutor<'e>,
220    ids: &[uuid::Uuid],
221) -> Result<u64, sqlx::Error> {
222    let result = sqlx::query!(
223        r#"UPDATE inbox SET dead_at = NULL, attempts = 0, updated_at = now()
224            WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
225        ids,
226    )
227    .execute(executor)
228    .await?;
229
230    Ok(result.rows_affected())
231}
232
233async fn purge_dead_rows<'e>(
234    executor: impl sqlx::PgExecutor<'e>,
235    ids: &[uuid::Uuid],
236) -> Result<u64, sqlx::Error> {
237    let result = sqlx::query!(
238        r#"DELETE FROM inbox WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
239        ids,
240    )
241    .execute(executor)
242    .await?;
243
244    Ok(result.rows_affected())
245}
reliar_store_postgres/inbox/dead_letters.rs

reliar_store_postgres/inbox/
dead_letters.rs
