reliar_store_postgres/outbox/
outbox_store_dead_letters.rs

1//! [`PostgresOutboxStore`]'s [`OutboxDeadLetters`] implementation: `list_dead`/`retry_dead`/
2//! `purge_dead`. The whole store-layer policy lives in this trait impl, reading `self.session`
3//! directly and calling `dead_letters`'s query functions for their statements.
4
5use sqlx::PgConnection;
6
7use reliar_core::Serializer;
8use reliar_outbox::{
9    DeadCursor, DeadLetterPage, DeadQuery, OutboxDeadLetters, OutboxRecordId, PoisonedRow,
10    RecordRef,
11};
12
13use super::dead_letters as repo;
14use super::error::PostgresOutboxError;
15use crate::records::{RawRow, decode_row};
16
17use super::PostgresOutboxStore;
18
19/// The largest `DeadQuery::limit` [`OutboxDeadLetters::list_dead`] honours — a caller-supplied
20/// value above this is silently capped, never sent to the database: this store is
21/// provider-capped, with a default of 100.
22const MAX_LIST_DEAD_LIMIT: u32 = 1000;
23
24impl<Ser: Serializer + Send + Sync + 'static> OutboxDeadLetters for PostgresOutboxStore<Ser> {
25    type Error = PostgresOutboxError;
26
27    /// **`ORDER BY dead_at ASC, id ASC` is normative**: `after` is a composite keyset
28    /// cursor over the columns `ix_outbox_dead_cursor` orders by; `message_type`/`tenant_id`/
29    /// `dead_before` are filters only. The cursor returned comes from the last row **scanned**,
30    /// poisoned rows included, so a poisoned tail cannot loop the caller forever.
31    async fn list_dead(&self, query: DeadQuery) -> Result<DeadLetterPage, Self::Error> {
32        // Provider-capped, default 100: a caller-supplied
33        // limit above this never reaches the database, regardless of what `DeadQuery` carries.
34        let capped_limit = query.limit.min(MAX_LIST_DEAD_LIMIT);
35        let limit = i64::from(capped_limit);
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
40        let rows: Vec<RawRow> = self
41            .session
42            .run(async |conn: &mut PgConnection| {
43                repo::list_dead_rows(
44                    &mut *conn,
45                    repo::ListDeadRowsParams {
46                        message_type: query.message_type.as_deref(),
47                        tenant_id: query.tenant_id.as_deref(),
48                        dead_before: query.dead_before,
49                        after_dead_at,
50                        after_id,
51                        limit,
52                    },
53                )
54                .await
55            })
56            .await
57            .map_err(|e| self.session.map_err::<PostgresOutboxError>(e))?;
58
59        let scanned = rows.len();
60        let mut records = Vec::with_capacity(scanned);
61        let mut poisoned = Vec::new();
62        let mut last_cursor: Option<DeadCursor> = None;
63
64        for raw in rows {
65            debug_assert!(raw.dead_at.is_some(), "list_dead selects only dead rows");
66
67            if let Some(dead_at) = raw.dead_at {
68                last_cursor = Some(DeadCursor::new(dead_at, OutboxRecordId::from_uuid(raw.id)));
69            }
70
71            match decode_row(raw) {
72                Ok(record) => records.push(record),
73                Err(err) => poisoned.push(PoisonedRow::new(err.id, err.message_id, err.detail)),
74            }
75        }
76
77        // "Full" is scanned == limit, poisoned rows included — they occupy a row in the scan,
78        // so counting only decoded records would stop pagination early on a poisoned tail.
79        let next_after = if scanned == capped_limit as usize {
80            last_cursor
81        } else {
82            None
83        };
84
85        Ok(DeadLetterPage::new(records, poisoned, next_after))
86    }
87
88    /// Returns dead rows to pending: clears the lease that already isn't there, resets
89    /// `attempts` to 0 (the **only** operation that does), keeps `last_error` for audit. Not
90    /// guarded by `claim_token` at all — a dead row holds no claim, so there is nothing to check
91    /// against (ADR 0046 Amendment A). Also clears `claim_token` on the resurrected row: the row
92    /// died holding whatever token its last claim stamped, and a stale outcome write from that
93    /// pre-death claim must not be able to match the row it resurrects into.
94    async fn retry_dead(&self, refs: &[RecordRef]) -> Result<u64, Self::Error> {
95        if refs.is_empty() {
96            return Ok(0);
97        }
98
99        let ids: Vec<uuid::Uuid> = refs.iter().map(|r| r.id.as_uuid()).collect();
100        let affected = self
101            .session
102            .run(async |conn: &mut PgConnection| {
103                repo::retry_dead_rows(&mut *conn, repo::RetryDeadRowsParams { ids: &ids }).await
104            })
105            .await
106            .map_err(|e| self.session.map_err::<PostgresOutboxError>(e))?;
107
108        Ok(affected)
109    }
110
111    /// Deletes dead rows by reference, regardless of
112    /// [`PurgeRequest::dead_retention`](reliar_outbox::PurgeRequest::dead_retention).
113    async fn purge_dead(&self, refs: &[RecordRef]) -> Result<u64, Self::Error> {
114        if refs.is_empty() {
115            return Ok(0);
116        }
117
118        let ids: Vec<uuid::Uuid> = refs.iter().map(|r| r.id.as_uuid()).collect();
119        let affected = self
120            .session
121            .run(async |conn: &mut PgConnection| {
122                repo::purge_dead_rows(&mut *conn, repo::PurgeDeadRowsParams { ids: &ids }).await
123            })
124            .await
125            .map_err(|e| self.session.map_err::<PostgresOutboxError>(e))?;
126
127        Ok(affected)
128    }
129}
reliar_store_postgres/outbox/outbox_store_dead_letters.rs

reliar_store_postgres/outbox/
outbox_store_dead_letters.rs
