mailrs_outbound_queue/queue/
lifecycle.rs

1//! A message's path through the queue: claim, then one of the four
2//! terminal marks, plus the sweep that reclaims a crashed claim.
3
4use super::*;
5#[cfg(feature = "pg")]
6use crate::BackendPool;
7
8/// fetch pending messages ready for delivery (legacy, non-atomic).
9///
10/// Returns rows in `status = 'pending'` without locking or marking
11/// them. Multi-worker setups using this then-`mark_inflight` flow can
12/// race on the same row and deliver twice. Prefer
13/// [`claim_for_delivery`] for any worker that may run with siblings.
14/// Kept for callers that only need a read snapshot.
15#[cfg(feature = "pg")]
16pub async fn dequeue(
17    pool: &BackendPool,
18    now: i64,
19    limit: u32,
20) -> Result<Vec<QueuedMessage>, sqlx::Error> {
21    #[allow(clippy::type_complexity)]
22    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
23        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
24         FROM outbound_queue
25         WHERE status = 'pending' AND next_retry <= $1
26         ORDER BY next_retry ASC
27         LIMIT $2",
28    )
29    .bind(now)
30    .bind(limit as i32)
31    .fetch_all(pool)
32    .await?;
33
34    Ok(rows
35        .into_iter()
36        .map(|r| QueuedMessage {
37            id: r.0,
38            sender: r.1,
39            recipient: r.2,
40            domain: r.3,
41            message_data: r.4,
42            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
43            attempts: r.6 as u32,
44            max_attempts: r.7 as u32,
45            next_retry: r.8,
46            last_error: r.9,
47            message_id: r.10,
48            created_at: r.11,
49            updated_at: r.12,
50            is_forwarded: r.13,
51        })
52        .collect())
53}
54
55/// Atomically claim up to `limit` pending messages and transition
56/// them to `inflight`, returning the claimed rows.
57///
58/// Implemented as `UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP
59/// LOCKED) RETURNING …` so concurrent workers never pick the same
60/// row (SKIP LOCKED skips rows already locked by other workers),
61/// and the claim+transition collapse to a single round-trip and
62/// single WAL fsync per batch instead of one SELECT + N per-row
63/// UPDATEs.
64///
65/// Correctness vs the legacy `dequeue` + per-row `mark_inflight`
66/// flow: SKIP LOCKED prevents the duplicate-delivery race that the
67/// pre-existing flow exposed when more than one worker was running
68/// concurrently (both workers would `SELECT` the same pending rows
69/// and both proceed past their racing `UPDATE`s). With this claim
70/// path, each pending row is delivered by at most one worker.
71#[cfg(feature = "pg")]
72pub async fn claim_for_delivery(
73    pool: &BackendPool,
74    now: i64,
75    limit: u32,
76) -> Result<Vec<QueuedMessage>, sqlx::Error> {
77    #[allow(clippy::type_complexity)]
78    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
79        "UPDATE outbound_queue
80         SET status = 'inflight', updated_at = $1
81         WHERE id IN (
82           SELECT id FROM outbound_queue
83           WHERE status = 'pending' AND next_retry <= $2
84           ORDER BY next_retry ASC
85           LIMIT $3
86           FOR UPDATE SKIP LOCKED
87         )
88         RETURNING id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded",
89    )
90    .bind(now)
91    .bind(now)
92    .bind(limit as i32)
93    .fetch_all(pool)
94    .await?;
95
96    Ok(rows
97        .into_iter()
98        .map(|r| QueuedMessage {
99            id: r.0,
100            sender: r.1,
101            recipient: r.2,
102            domain: r.3,
103            message_data: r.4,
104            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::InFlight),
105            attempts: r.6 as u32,
106            max_attempts: r.7 as u32,
107            next_retry: r.8,
108            last_error: r.9,
109            message_id: r.10,
110            created_at: r.11,
111            updated_at: r.12,
112            is_forwarded: r.13,
113        })
114        .collect())
115}
116
117/// mark a message as in-flight
118#[cfg(feature = "pg")]
119pub async fn mark_inflight(pool: &BackendPool, id: i64, now: i64) -> Result<(), sqlx::Error> {
120    sqlx::query("UPDATE outbound_queue SET status = 'inflight', updated_at = $1 WHERE id = $2")
121        .bind(now)
122        .bind(id)
123        .execute(pool)
124        .await?;
125    Ok(())
126}
127
128/// mark a message as delivered
129#[cfg(feature = "pg")]
130pub async fn mark_delivered(pool: &BackendPool, id: i64, now: i64) -> Result<(), sqlx::Error> {
131    sqlx::query("UPDATE outbound_queue SET status = 'delivered', updated_at = $1 WHERE id = $2")
132        .bind(now)
133        .bind(id)
134        .execute(pool)
135        .await?;
136    Ok(())
137}
138
139/// mark a message as failed with next retry time
140#[cfg(feature = "pg")]
141pub async fn mark_failed(
142    pool: &BackendPool,
143    id: i64,
144    error: &str,
145    next_retry: i64,
146    now: i64,
147) -> Result<(), sqlx::Error> {
148    sqlx::query(
149        "UPDATE outbound_queue SET status = 'pending', attempts = attempts + 1, last_error = $1, next_retry = $2, updated_at = $3 WHERE id = $4",
150    )
151    .bind(error)
152    .bind(next_retry)
153    .bind(now)
154    .bind(id)
155    .execute(pool)
156    .await?;
157    Ok(())
158}
159
160/// mark a message as permanently bounced
161#[cfg(feature = "pg")]
162pub async fn mark_bounced(
163    pool: &BackendPool,
164    id: i64,
165    error: &str,
166    now: i64,
167) -> Result<(), sqlx::Error> {
168    sqlx::query(
169        "UPDATE outbound_queue SET status = 'bounced', last_error = $1, updated_at = $2 WHERE id = $3",
170    )
171    .bind(error)
172    .bind(now)
173    .bind(id)
174    .execute(pool)
175    .await?;
176    Ok(())
177}
178
179/// recover messages stuck in inflight status for more than 10 minutes
180/// (worker crashed or was killed before marking them as delivered/failed)
181#[cfg(feature = "pg")]
182pub async fn recover_stale_inflight(pool: &BackendPool, now: i64) -> Result<u64, sqlx::Error> {
183    let stale_threshold = now - 600; // 10 minutes
184    let result = sqlx::query(
185        "UPDATE outbound_queue SET status = 'pending', updated_at = $1 \
186         WHERE status = 'inflight' AND updated_at < $2",
187    )
188    .bind(now)
189    .bind(stale_threshold)
190    .execute(pool)
191    .await?;
192    Ok(result.rows_affected())
193}
mailrs_outbound_queue/queue/lifecycle.rs

mailrs_outbound_queue/queue/
lifecycle.rs
