mailrs_outbound_queue/
queue.rs

1#[cfg(feature = "pg")]
2use redis::AsyncCommands;
3#[cfg(feature = "pg")]
4use sqlx::PgPool;
5
6/// Lifecycle status of a queued message.
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub enum QueueStatus {
9    /// Awaiting first delivery attempt.
10    Pending,
11    /// Currently being delivered by a worker.
12    InFlight,
13    /// Successfully accepted by the remote MX.
14    Delivered,
15    /// Last attempt failed; will retry per backoff schedule.
16    Failed,
17    /// Permanent failure; will NOT retry. A DSN has been queued back.
18    Bounced,
19}
20
21impl QueueStatus {
22    /// Lower-snake-case rendering for SQL persistence.
23    pub fn as_str(&self) -> &'static str {
24        match self {
25            Self::Pending => "pending",
26            Self::InFlight => "inflight",
27            Self::Delivered => "delivered",
28            Self::Failed => "failed",
29            Self::Bounced => "bounced",
30        }
31    }
32
33    /// Inverse of [`Self::as_str`]; `None` on unknown values.
34    pub fn parse(s: &str) -> Option<Self> {
35        match s {
36            "pending" => Some(Self::Pending),
37            "inflight" => Some(Self::InFlight),
38            "delivered" => Some(Self::Delivered),
39            "failed" => Some(Self::Failed),
40            "bounced" => Some(Self::Bounced),
41            _ => None,
42        }
43    }
44}
45
46/// One queued outbound message — the full row stored in the outbound queue.
47#[derive(Debug, Clone)]
48pub struct QueuedMessage {
49    /// Store-native primary key.
50    pub id: i64,
51    /// Envelope sender (reverse path).
52    pub sender: String,
53    /// Envelope recipient (single forward path; multi-recipient messages
54    /// fan out into one row per recipient).
55    pub recipient: String,
56    /// Recipient's domain — extracted for MX-grouped batching.
57    pub domain: String,
58    /// Full RFC 5322 message body (including headers).
59    pub message_data: Vec<u8>,
60    /// Current lifecycle status.
61    pub status: QueueStatus,
62    /// Number of delivery attempts made so far.
63    pub attempts: u32,
64    /// Cap after which `Failed` flips to `Bounced`.
65    pub max_attempts: u32,
66    /// Epoch seconds — the earliest time the next attempt is eligible.
67    pub next_retry: i64,
68    /// Last error response from the remote MX, if any.
69    pub last_error: Option<String>,
70    /// `Message-ID:` header value, for log correlation.
71    pub message_id: Option<String>,
72    /// Epoch seconds when the row was first enqueued.
73    pub created_at: i64,
74    /// Epoch seconds of the most recent update.
75    pub updated_at: i64,
76    /// `true` when the message came from a forwarding rule rather than a
77    /// local sender.
78    pub is_forwarded: bool,
79}
80
81/// enqueue a message for outbound delivery
82#[cfg(feature = "pg")]
83pub async fn enqueue(
84    pool: &PgPool,
85    sender: &str,
86    recipient: &str,
87    domain: &str,
88    message_data: &[u8],
89    message_id: Option<&str>,
90    now: i64,
91) -> Result<i64, sqlx::Error> {
92    enqueue_ex(
93        pool,
94        sender,
95        recipient,
96        domain,
97        message_data,
98        message_id,
99        now,
100        false,
101    )
102    .await
103}
104
105/// enqueue a message for outbound delivery with forwarding flag
106#[allow(clippy::too_many_arguments)]
107#[cfg(feature = "pg")]
108pub async fn enqueue_ex(
109    pool: &PgPool,
110    sender: &str,
111    recipient: &str,
112    domain: &str,
113    message_data: &[u8],
114    message_id: Option<&str>,
115    now: i64,
116    is_forwarded: bool,
117) -> Result<i64, sqlx::Error> {
118    let row: (i64,) = sqlx::query_as(
119        "INSERT INTO outbound_queue (sender, recipient, domain, message_data, status, next_retry, message_id, created_at, updated_at, is_forwarded)
120         VALUES ($1, $2, $3, $4, 'pending', $5, $6, $5, $5, $7)
121         RETURNING id",
122    )
123    .bind(sender)
124    .bind(recipient)
125    .bind(domain)
126    .bind(message_data)
127    .bind(now)
128    .bind(message_id)
129    .bind(is_forwarded)
130    .fetch_one(pool)
131    .await?;
132    Ok(row.0)
133}
134
135/// notify the delivery worker that new messages are queued
136#[cfg(feature = "pg")]
137pub async fn notify(valkey: &mut redis::aio::ConnectionManager) {
138    let _: Result<i32, _> = valkey.publish("queue:notify", "1").await;
139}
140
141/// recover messages stuck in inflight status for more than 10 minutes
142/// (worker crashed or was killed before marking them as delivered/failed)
143#[cfg(feature = "pg")]
144pub async fn recover_stale_inflight(pool: &PgPool, now: i64) -> Result<u64, sqlx::Error> {
145    let stale_threshold = now - 600; // 10 minutes
146    let result = sqlx::query(
147        "UPDATE outbound_queue SET status = 'pending', updated_at = $1 \
148         WHERE status = 'inflight' AND updated_at < $2",
149    )
150    .bind(now)
151    .bind(stale_threshold)
152    .execute(pool)
153    .await?;
154    Ok(result.rows_affected())
155}
156
157/// Count of rows in `status = 'pending'` (any `next_retry`). Used by
158/// the delivery worker to publish a `mailrs_outbound_queue_depth` gauge
159/// per poll tick. O(rows) but cheap with the `status` index.
160#[cfg(feature = "pg")]
161pub async fn count_pending(pool: &PgPool) -> Result<i64, sqlx::Error> {
162    let (n,): (i64,) =
163        sqlx::query_as("SELECT count(*) FROM outbound_queue WHERE status = 'pending'")
164            .fetch_one(pool)
165            .await?;
166    Ok(n)
167}
168
169/// Count of rows in `status = 'inflight'` (currently being delivered
170/// by a worker). Used alongside `count_pending` so dashboards can show
171/// both "queued" and "in-flight" depth.
172#[cfg(feature = "pg")]
173pub async fn count_inflight(pool: &PgPool) -> Result<i64, sqlx::Error> {
174    let (n,): (i64,) =
175        sqlx::query_as("SELECT count(*) FROM outbound_queue WHERE status = 'inflight'")
176            .fetch_one(pool)
177            .await?;
178    Ok(n)
179}
180
181/// fetch pending messages ready for delivery (legacy, non-atomic).
182///
183/// Returns rows in `status = 'pending'` without locking or marking
184/// them. Multi-worker setups using this then-`mark_inflight` flow can
185/// race on the same row and deliver twice. Prefer
186/// [`claim_for_delivery`] for any worker that may run with siblings.
187/// Kept for callers that only need a read snapshot.
188#[cfg(feature = "pg")]
189pub async fn dequeue(
190    pool: &PgPool,
191    now: i64,
192    limit: u32,
193) -> Result<Vec<QueuedMessage>, sqlx::Error> {
194    #[allow(clippy::type_complexity)]
195    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
196        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
197         FROM outbound_queue
198         WHERE status = 'pending' AND next_retry <= $1
199         ORDER BY next_retry ASC
200         LIMIT $2",
201    )
202    .bind(now)
203    .bind(limit as i32)
204    .fetch_all(pool)
205    .await?;
206
207    Ok(rows
208        .into_iter()
209        .map(|r| QueuedMessage {
210            id: r.0,
211            sender: r.1,
212            recipient: r.2,
213            domain: r.3,
214            message_data: r.4,
215            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
216            attempts: r.6 as u32,
217            max_attempts: r.7 as u32,
218            next_retry: r.8,
219            last_error: r.9,
220            message_id: r.10,
221            created_at: r.11,
222            updated_at: r.12,
223            is_forwarded: r.13,
224        })
225        .collect())
226}
227
228/// Atomically claim up to `limit` pending messages and transition
229/// them to `inflight`, returning the claimed rows.
230///
231/// Implemented as `UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP
232/// LOCKED) RETURNING …` so concurrent workers never pick the same
233/// row (SKIP LOCKED skips rows already locked by other workers),
234/// and the claim+transition collapse to a single round-trip and
235/// single WAL fsync per batch instead of one SELECT + N per-row
236/// UPDATEs.
237///
238/// Correctness vs the legacy `dequeue` + per-row `mark_inflight`
239/// flow: SKIP LOCKED prevents the duplicate-delivery race that the
240/// pre-existing flow exposed when more than one worker was running
241/// concurrently (both workers would `SELECT` the same pending rows
242/// and both proceed past their racing `UPDATE`s). With this claim
243/// path, each pending row is delivered by at most one worker.
244#[cfg(feature = "pg")]
245pub async fn claim_for_delivery(
246    pool: &PgPool,
247    now: i64,
248    limit: u32,
249) -> Result<Vec<QueuedMessage>, sqlx::Error> {
250    #[allow(clippy::type_complexity)]
251    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
252        "UPDATE outbound_queue
253         SET status = 'inflight', updated_at = $1
254         WHERE id IN (
255           SELECT id FROM outbound_queue
256           WHERE status = 'pending' AND next_retry <= $2
257           ORDER BY next_retry ASC
258           LIMIT $3
259           FOR UPDATE SKIP LOCKED
260         )
261         RETURNING id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded",
262    )
263    .bind(now)
264    .bind(now)
265    .bind(limit as i32)
266    .fetch_all(pool)
267    .await?;
268
269    Ok(rows
270        .into_iter()
271        .map(|r| QueuedMessage {
272            id: r.0,
273            sender: r.1,
274            recipient: r.2,
275            domain: r.3,
276            message_data: r.4,
277            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::InFlight),
278            attempts: r.6 as u32,
279            max_attempts: r.7 as u32,
280            next_retry: r.8,
281            last_error: r.9,
282            message_id: r.10,
283            created_at: r.11,
284            updated_at: r.12,
285            is_forwarded: r.13,
286        })
287        .collect())
288}
289
290/// mark a message as in-flight
291#[cfg(feature = "pg")]
292pub async fn mark_inflight(pool: &PgPool, id: i64, now: i64) -> Result<(), sqlx::Error> {
293    sqlx::query("UPDATE outbound_queue SET status = 'inflight', updated_at = $1 WHERE id = $2")
294        .bind(now)
295        .bind(id)
296        .execute(pool)
297        .await?;
298    Ok(())
299}
300
301/// mark a message as delivered
302#[cfg(feature = "pg")]
303pub async fn mark_delivered(pool: &PgPool, id: i64, now: i64) -> Result<(), sqlx::Error> {
304    sqlx::query("UPDATE outbound_queue SET status = 'delivered', updated_at = $1 WHERE id = $2")
305        .bind(now)
306        .bind(id)
307        .execute(pool)
308        .await?;
309    Ok(())
310}
311
312/// mark a message as failed with next retry time
313#[cfg(feature = "pg")]
314pub async fn mark_failed(
315    pool: &PgPool,
316    id: i64,
317    error: &str,
318    next_retry: i64,
319    now: i64,
320) -> Result<(), sqlx::Error> {
321    sqlx::query(
322        "UPDATE outbound_queue SET status = 'pending', attempts = attempts + 1, last_error = $1, next_retry = $2, updated_at = $3 WHERE id = $4",
323    )
324    .bind(error)
325    .bind(next_retry)
326    .bind(now)
327    .bind(id)
328    .execute(pool)
329    .await?;
330    Ok(())
331}
332
333/// mark a message as permanently bounced
334#[cfg(feature = "pg")]
335pub async fn mark_bounced(
336    pool: &PgPool,
337    id: i64,
338    error: &str,
339    now: i64,
340) -> Result<(), sqlx::Error> {
341    sqlx::query(
342        "UPDATE outbound_queue SET status = 'bounced', last_error = $1, updated_at = $2 WHERE id = $3",
343    )
344    .bind(error)
345    .bind(now)
346    .bind(id)
347    .execute(pool)
348    .await?;
349    Ok(())
350}
351
352/// get queue statistics
353#[cfg(feature = "pg")]
354pub async fn queue_stats(pool: &PgPool) -> Result<Vec<(String, i64)>, sqlx::Error> {
355    let rows: Vec<(String, i64)> =
356        sqlx::query_as("SELECT status, COUNT(*) FROM outbound_queue GROUP BY status")
357            .fetch_all(pool)
358            .await?;
359    Ok(rows)
360}
361
362/// get a specific queued message by id
363#[cfg(feature = "pg")]
364pub async fn get_message(pool: &PgPool, id: i64) -> Result<Option<QueuedMessage>, sqlx::Error> {
365    #[allow(clippy::type_complexity)]
366    let row: Option<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
367        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
368         FROM outbound_queue WHERE id = $1",
369    )
370    .bind(id)
371    .fetch_optional(pool)
372    .await?;
373
374    Ok(row.map(|r| QueuedMessage {
375        id: r.0,
376        sender: r.1,
377        recipient: r.2,
378        domain: r.3,
379        message_data: r.4,
380        status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
381        attempts: r.6 as u32,
382        max_attempts: r.7 as u32,
383        next_retry: r.8,
384        last_error: r.9,
385        message_id: r.10,
386        created_at: r.11,
387        updated_at: r.12,
388        is_forwarded: r.13,
389    }))
390}
391
392/// enqueue a message for scheduled delivery at a future time
393#[allow(clippy::too_many_arguments)]
394#[cfg(feature = "pg")]
395pub async fn enqueue_scheduled(
396    pool: &PgPool,
397    sender: &str,
398    recipient: &str,
399    domain: &str,
400    message_data: &[u8],
401    message_id: Option<&str>,
402    created_at: i64,
403    scheduled_at: i64,
404) -> Result<i64, sqlx::Error> {
405    let row: (i64,) = sqlx::query_as(
406        "INSERT INTO outbound_queue (sender, recipient, domain, message_data, status, next_retry, message_id, created_at, updated_at, is_forwarded)
407         VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7, $7, false)
408         RETURNING id",
409    )
410    .bind(sender)
411    .bind(recipient)
412    .bind(domain)
413    .bind(message_data)
414    .bind(scheduled_at)
415    .bind(message_id)
416    .bind(created_at)
417    .fetch_one(pool)
418    .await?;
419    Ok(row.0)
420}
421
422/// cancel a pending outbound message (undo send)
423#[cfg(feature = "pg")]
424pub async fn cancel_pending(pool: &PgPool, id: i64) -> Result<bool, sqlx::Error> {
425    let result = sqlx::query("DELETE FROM outbound_queue WHERE id = $1 AND status = 'pending'")
426        .bind(id)
427        .execute(pool)
428        .await?;
429    Ok(result.rows_affected() > 0)
430}
431
432/// cancel a pending outbound message by message_id (undo send)
433#[cfg(feature = "pg")]
434pub async fn cancel_pending_by_message_id(
435    pool: &PgPool,
436    message_id: &str,
437    sender: &str,
438) -> Result<bool, sqlx::Error> {
439    let result = sqlx::query(
440        "DELETE FROM outbound_queue WHERE message_id = $1 AND status = 'pending' AND sender = $2",
441    )
442    .bind(message_id)
443    .bind(sender)
444    .execute(pool)
445    .await?;
446    Ok(result.rows_affected() > 0)
447}
448
449/// reset a bounced/failed message back to pending for retry
450#[cfg(feature = "pg")]
451pub async fn retry_message(pool: &PgPool, id: i64, now: i64) -> Result<bool, sqlx::Error> {
452    let result = sqlx::query(
453        "UPDATE outbound_queue SET status = 'pending', next_retry = $1, updated_at = $1 WHERE id = $2 AND status IN ('bounced', 'failed')",
454    )
455    .bind(now)
456    .bind(id)
457    .execute(pool)
458    .await?;
459    Ok(result.rows_affected() > 0)
460}
461
462/// list recent queue entries for admin UI
463#[cfg(feature = "pg")]
464pub async fn list_recent(pool: &PgPool, limit: i32) -> Result<Vec<QueuedMessage>, sqlx::Error> {
465    #[allow(clippy::type_complexity)]
466    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
467        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
468         FROM outbound_queue
469         ORDER BY created_at DESC
470         LIMIT $1",
471    )
472    .bind(limit)
473    .fetch_all(pool)
474    .await?;
475
476    Ok(rows
477        .into_iter()
478        .map(|r| QueuedMessage {
479            id: r.0,
480            sender: r.1,
481            recipient: r.2,
482            domain: r.3,
483            message_data: r.4,
484            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
485            attempts: r.6 as u32,
486            max_attempts: r.7 as u32,
487            next_retry: r.8,
488            last_error: r.9,
489            message_id: r.10,
490            created_at: r.11,
491            updated_at: r.12,
492            is_forwarded: r.13,
493        })
494        .collect())
495}
496
497mod suppression;
498
499#[cfg(feature = "pg")]
500pub use suppression::{add_suppression, is_suppressed, list_suppressions, remove_suppression};
501pub use suppression::is_hard_bounce;
502
503#[cfg(test)]
504mod tests {
505    use super::*;
506
507    #[test]
508    fn queue_status_roundtrip() {
509        let variants = [
510            QueueStatus::Pending,
511            QueueStatus::InFlight,
512            QueueStatus::Delivered,
513            QueueStatus::Failed,
514            QueueStatus::Bounced,
515        ];
516        for v in &variants {
517            let s = v.as_str();
518            let parsed = QueueStatus::parse(s).unwrap();
519            assert_eq!(&parsed, v, "roundtrip failed for {s}");
520        }
521    }
522
523    #[test]
524    fn queue_status_parse_unknown() {
525        assert_eq!(QueueStatus::parse("unknown"), None);
526        assert_eq!(QueueStatus::parse(""), None);
527        assert_eq!(QueueStatus::parse("PENDING"), None);
528    }
529
530    #[test]
531    fn queue_status_as_str_values() {
532        assert_eq!(QueueStatus::Pending.as_str(), "pending");
533        assert_eq!(QueueStatus::InFlight.as_str(), "inflight");
534        assert_eq!(QueueStatus::Delivered.as_str(), "delivered");
535        assert_eq!(QueueStatus::Failed.as_str(), "failed");
536        assert_eq!(QueueStatus::Bounced.as_str(), "bounced");
537    }
538
539    #[test]
540    fn queue_status_parse_case_sensitive() {
541        // parse is case-sensitive — uppercase variants are not valid
542        assert_eq!(QueueStatus::parse("Pending"), None);
543        assert_eq!(QueueStatus::parse("InFlight"), None);
544        assert_eq!(QueueStatus::parse("DELIVERED"), None);
545        assert_eq!(QueueStatus::parse("Failed"), None);
546        assert_eq!(QueueStatus::parse("Bounced"), None);
547    }
548
549    #[test]
550    fn queue_status_parse_whitespace_rejected() {
551        assert_eq!(QueueStatus::parse(" pending"), None);
552        assert_eq!(QueueStatus::parse("pending "), None);
553        assert_eq!(QueueStatus::parse("  "), None);
554    }
555
556    #[test]
557    fn queue_status_eq() {
558        assert_eq!(QueueStatus::Pending, QueueStatus::Pending);
559        assert_ne!(QueueStatus::Pending, QueueStatus::Delivered);
560        assert_ne!(QueueStatus::Failed, QueueStatus::Bounced);
561    }
562
563    #[test]
564    fn queue_status_clone() {
565        let s = QueueStatus::InFlight;
566        let c = s.clone();
567        assert_eq!(s, c);
568    }
569
570    #[test]
571    fn queued_message_clone_preserves_fields() {
572        let msg = QueuedMessage {
573            id: 42,
574            sender: "s@example.com".into(),
575            recipient: "r@remote.com".into(),
576            domain: "remote.com".into(),
577            message_data: vec![1, 2, 3],
578            status: QueueStatus::Pending,
579            attempts: 3,
580            max_attempts: 8,
581            next_retry: 1_700_000_000,
582            last_error: Some("temporary failure".into()),
583            message_id: Some("msg-id-123".into()),
584            created_at: 1_699_000_000,
585            updated_at: 1_699_500_000,
586            is_forwarded: true,
587        };
588        let cloned = msg.clone();
589        assert_eq!(cloned.id, 42);
590        assert_eq!(cloned.sender, "s@example.com");
591        assert_eq!(cloned.recipient, "r@remote.com");
592        assert_eq!(cloned.domain, "remote.com");
593        assert_eq!(cloned.message_data, vec![1, 2, 3]);
594        assert_eq!(cloned.attempts, 3);
595        assert_eq!(cloned.max_attempts, 8);
596        assert_eq!(cloned.next_retry, 1_700_000_000);
597        assert_eq!(cloned.last_error, Some("temporary failure".into()));
598        assert_eq!(cloned.message_id, Some("msg-id-123".into()));
599        assert!(cloned.is_forwarded);
600    }
601
602    #[test]
603    fn queued_message_no_last_error() {
604        let msg = QueuedMessage {
605            id: 1,
606            sender: "s@example.com".into(),
607            recipient: "r@remote.com".into(),
608            domain: "remote.com".into(),
609            message_data: vec![],
610            status: QueueStatus::Pending,
611            attempts: 0,
612            max_attempts: 8,
613            next_retry: 0,
614            last_error: None,
615            message_id: None,
616            created_at: 0,
617            updated_at: 0,
618            is_forwarded: false,
619        };
620        assert!(msg.last_error.is_none());
621        assert!(msg.message_id.is_none());
622    }
623}
mailrs_outbound_queue/queue.rs

mailrs_outbound_queue/
queue.rs
