mailrs_outbound_queue/
queue.rs

1#[cfg(feature = "pg")]
2use kevy_embedded::Store;
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 via the
136/// in-process kevy `queue:notify` channel.
137#[cfg(feature = "pg")]
138pub fn notify(kevy: &Store) {
139    let _ = kevy.publish(b"queue:notify", b"1");
140}
141
142/// recover messages stuck in inflight status for more than 10 minutes
143/// (worker crashed or was killed before marking them as delivered/failed)
144#[cfg(feature = "pg")]
145pub async fn recover_stale_inflight(pool: &PgPool, now: i64) -> Result<u64, sqlx::Error> {
146    let stale_threshold = now - 600; // 10 minutes
147    let result = sqlx::query(
148        "UPDATE outbound_queue SET status = 'pending', updated_at = $1 \
149         WHERE status = 'inflight' AND updated_at < $2",
150    )
151    .bind(now)
152    .bind(stale_threshold)
153    .execute(pool)
154    .await?;
155    Ok(result.rows_affected())
156}
157
158/// Count of rows in `status = 'pending'` (any `next_retry`). Used by
159/// the delivery worker to publish a `mailrs_outbound_queue_depth` gauge
160/// per poll tick. O(rows) but cheap with the `status` index.
161#[cfg(feature = "pg")]
162pub async fn count_pending(pool: &PgPool) -> Result<i64, sqlx::Error> {
163    let (n,): (i64,) =
164        sqlx::query_as("SELECT count(*) FROM outbound_queue WHERE status = 'pending'")
165            .fetch_one(pool)
166            .await?;
167    Ok(n)
168}
169
170/// Count of rows in `status = 'inflight'` (currently being delivered
171/// by a worker). Used alongside `count_pending` so dashboards can show
172/// both "queued" and "in-flight" depth.
173#[cfg(feature = "pg")]
174pub async fn count_inflight(pool: &PgPool) -> Result<i64, sqlx::Error> {
175    let (n,): (i64,) =
176        sqlx::query_as("SELECT count(*) FROM outbound_queue WHERE status = 'inflight'")
177            .fetch_one(pool)
178            .await?;
179    Ok(n)
180}
181
182/// fetch pending messages ready for delivery (legacy, non-atomic).
183///
184/// Returns rows in `status = 'pending'` without locking or marking
185/// them. Multi-worker setups using this then-`mark_inflight` flow can
186/// race on the same row and deliver twice. Prefer
187/// [`claim_for_delivery`] for any worker that may run with siblings.
188/// Kept for callers that only need a read snapshot.
189#[cfg(feature = "pg")]
190pub async fn dequeue(
191    pool: &PgPool,
192    now: i64,
193    limit: u32,
194) -> Result<Vec<QueuedMessage>, sqlx::Error> {
195    #[allow(clippy::type_complexity)]
196    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
197        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
198         FROM outbound_queue
199         WHERE status = 'pending' AND next_retry <= $1
200         ORDER BY next_retry ASC
201         LIMIT $2",
202    )
203    .bind(now)
204    .bind(limit as i32)
205    .fetch_all(pool)
206    .await?;
207
208    Ok(rows
209        .into_iter()
210        .map(|r| QueuedMessage {
211            id: r.0,
212            sender: r.1,
213            recipient: r.2,
214            domain: r.3,
215            message_data: r.4,
216            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
217            attempts: r.6 as u32,
218            max_attempts: r.7 as u32,
219            next_retry: r.8,
220            last_error: r.9,
221            message_id: r.10,
222            created_at: r.11,
223            updated_at: r.12,
224            is_forwarded: r.13,
225        })
226        .collect())
227}
228
229/// Atomically claim up to `limit` pending messages and transition
230/// them to `inflight`, returning the claimed rows.
231///
232/// Implemented as `UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP
233/// LOCKED) RETURNING …` so concurrent workers never pick the same
234/// row (SKIP LOCKED skips rows already locked by other workers),
235/// and the claim+transition collapse to a single round-trip and
236/// single WAL fsync per batch instead of one SELECT + N per-row
237/// UPDATEs.
238///
239/// Correctness vs the legacy `dequeue` + per-row `mark_inflight`
240/// flow: SKIP LOCKED prevents the duplicate-delivery race that the
241/// pre-existing flow exposed when more than one worker was running
242/// concurrently (both workers would `SELECT` the same pending rows
243/// and both proceed past their racing `UPDATE`s). With this claim
244/// path, each pending row is delivered by at most one worker.
245#[cfg(feature = "pg")]
246pub async fn claim_for_delivery(
247    pool: &PgPool,
248    now: i64,
249    limit: u32,
250) -> Result<Vec<QueuedMessage>, sqlx::Error> {
251    #[allow(clippy::type_complexity)]
252    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
253        "UPDATE outbound_queue
254         SET status = 'inflight', updated_at = $1
255         WHERE id IN (
256           SELECT id FROM outbound_queue
257           WHERE status = 'pending' AND next_retry <= $2
258           ORDER BY next_retry ASC
259           LIMIT $3
260           FOR UPDATE SKIP LOCKED
261         )
262         RETURNING id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded",
263    )
264    .bind(now)
265    .bind(now)
266    .bind(limit as i32)
267    .fetch_all(pool)
268    .await?;
269
270    Ok(rows
271        .into_iter()
272        .map(|r| QueuedMessage {
273            id: r.0,
274            sender: r.1,
275            recipient: r.2,
276            domain: r.3,
277            message_data: r.4,
278            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::InFlight),
279            attempts: r.6 as u32,
280            max_attempts: r.7 as u32,
281            next_retry: r.8,
282            last_error: r.9,
283            message_id: r.10,
284            created_at: r.11,
285            updated_at: r.12,
286            is_forwarded: r.13,
287        })
288        .collect())
289}
290
291/// mark a message as in-flight
292#[cfg(feature = "pg")]
293pub async fn mark_inflight(pool: &PgPool, id: i64, now: i64) -> Result<(), sqlx::Error> {
294    sqlx::query("UPDATE outbound_queue SET status = 'inflight', updated_at = $1 WHERE id = $2")
295        .bind(now)
296        .bind(id)
297        .execute(pool)
298        .await?;
299    Ok(())
300}
301
302/// mark a message as delivered
303#[cfg(feature = "pg")]
304pub async fn mark_delivered(pool: &PgPool, id: i64, now: i64) -> Result<(), sqlx::Error> {
305    sqlx::query("UPDATE outbound_queue SET status = 'delivered', updated_at = $1 WHERE id = $2")
306        .bind(now)
307        .bind(id)
308        .execute(pool)
309        .await?;
310    Ok(())
311}
312
313/// mark a message as failed with next retry time
314#[cfg(feature = "pg")]
315pub async fn mark_failed(
316    pool: &PgPool,
317    id: i64,
318    error: &str,
319    next_retry: i64,
320    now: i64,
321) -> Result<(), sqlx::Error> {
322    sqlx::query(
323        "UPDATE outbound_queue SET status = 'pending', attempts = attempts + 1, last_error = $1, next_retry = $2, updated_at = $3 WHERE id = $4",
324    )
325    .bind(error)
326    .bind(next_retry)
327    .bind(now)
328    .bind(id)
329    .execute(pool)
330    .await?;
331    Ok(())
332}
333
334/// mark a message as permanently bounced
335#[cfg(feature = "pg")]
336pub async fn mark_bounced(
337    pool: &PgPool,
338    id: i64,
339    error: &str,
340    now: i64,
341) -> Result<(), sqlx::Error> {
342    sqlx::query(
343        "UPDATE outbound_queue SET status = 'bounced', last_error = $1, updated_at = $2 WHERE id = $3",
344    )
345    .bind(error)
346    .bind(now)
347    .bind(id)
348    .execute(pool)
349    .await?;
350    Ok(())
351}
352
353/// get queue statistics
354#[cfg(feature = "pg")]
355pub async fn queue_stats(pool: &PgPool) -> Result<Vec<(String, i64)>, sqlx::Error> {
356    let rows: Vec<(String, i64)> =
357        sqlx::query_as("SELECT status, COUNT(*) FROM outbound_queue GROUP BY status")
358            .fetch_all(pool)
359            .await?;
360    Ok(rows)
361}
362
363/// get a specific queued message by id
364#[cfg(feature = "pg")]
365pub async fn get_message(pool: &PgPool, id: i64) -> Result<Option<QueuedMessage>, sqlx::Error> {
366    #[allow(clippy::type_complexity)]
367    let row: Option<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
368        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
369         FROM outbound_queue WHERE id = $1",
370    )
371    .bind(id)
372    .fetch_optional(pool)
373    .await?;
374
375    Ok(row.map(|r| QueuedMessage {
376        id: r.0,
377        sender: r.1,
378        recipient: r.2,
379        domain: r.3,
380        message_data: r.4,
381        status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
382        attempts: r.6 as u32,
383        max_attempts: r.7 as u32,
384        next_retry: r.8,
385        last_error: r.9,
386        message_id: r.10,
387        created_at: r.11,
388        updated_at: r.12,
389        is_forwarded: r.13,
390    }))
391}
392
393/// enqueue a message for scheduled delivery at a future time
394#[allow(clippy::too_many_arguments)]
395#[cfg(feature = "pg")]
396pub async fn enqueue_scheduled(
397    pool: &PgPool,
398    sender: &str,
399    recipient: &str,
400    domain: &str,
401    message_data: &[u8],
402    message_id: Option<&str>,
403    created_at: i64,
404    scheduled_at: i64,
405) -> Result<i64, sqlx::Error> {
406    let row: (i64,) = sqlx::query_as(
407        "INSERT INTO outbound_queue (sender, recipient, domain, message_data, status, next_retry, message_id, created_at, updated_at, is_forwarded)
408         VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7, $7, false)
409         RETURNING id",
410    )
411    .bind(sender)
412    .bind(recipient)
413    .bind(domain)
414    .bind(message_data)
415    .bind(scheduled_at)
416    .bind(message_id)
417    .bind(created_at)
418    .fetch_one(pool)
419    .await?;
420    Ok(row.0)
421}
422
423/// cancel a pending outbound message (undo send)
424#[cfg(feature = "pg")]
425pub async fn cancel_pending(pool: &PgPool, id: i64) -> Result<bool, sqlx::Error> {
426    let result = sqlx::query("DELETE FROM outbound_queue WHERE id = $1 AND status = 'pending'")
427        .bind(id)
428        .execute(pool)
429        .await?;
430    Ok(result.rows_affected() > 0)
431}
432
433/// cancel a pending outbound message by message_id (undo send)
434#[cfg(feature = "pg")]
435pub async fn cancel_pending_by_message_id(
436    pool: &PgPool,
437    message_id: &str,
438    sender: &str,
439) -> Result<bool, sqlx::Error> {
440    let result = sqlx::query(
441        "DELETE FROM outbound_queue WHERE message_id = $1 AND status = 'pending' AND sender = $2",
442    )
443    .bind(message_id)
444    .bind(sender)
445    .execute(pool)
446    .await?;
447    Ok(result.rows_affected() > 0)
448}
449
450/// reset a bounced/failed message back to pending for retry
451#[cfg(feature = "pg")]
452pub async fn retry_message(pool: &PgPool, id: i64, now: i64) -> Result<bool, sqlx::Error> {
453    let result = sqlx::query(
454        "UPDATE outbound_queue SET status = 'pending', next_retry = $1, updated_at = $1 WHERE id = $2 AND status IN ('bounced', 'failed')",
455    )
456    .bind(now)
457    .bind(id)
458    .execute(pool)
459    .await?;
460    Ok(result.rows_affected() > 0)
461}
462
463/// list recent queue entries for admin UI
464#[cfg(feature = "pg")]
465pub async fn list_recent(pool: &PgPool, limit: i32) -> Result<Vec<QueuedMessage>, sqlx::Error> {
466    #[allow(clippy::type_complexity)]
467    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
468        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
469         FROM outbound_queue
470         ORDER BY created_at DESC
471         LIMIT $1",
472    )
473    .bind(limit)
474    .fetch_all(pool)
475    .await?;
476
477    Ok(rows
478        .into_iter()
479        .map(|r| QueuedMessage {
480            id: r.0,
481            sender: r.1,
482            recipient: r.2,
483            domain: r.3,
484            message_data: r.4,
485            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
486            attempts: r.6 as u32,
487            max_attempts: r.7 as u32,
488            next_retry: r.8,
489            last_error: r.9,
490            message_id: r.10,
491            created_at: r.11,
492            updated_at: r.12,
493            is_forwarded: r.13,
494        })
495        .collect())
496}
497
498mod suppression;
499
500pub use suppression::is_hard_bounce;
501#[cfg(feature = "pg")]
502pub use suppression::{add_suppression, is_suppressed, list_suppressions, remove_suppression};
503
504#[cfg(test)]
505mod tests {
506    use super::*;
507
508    #[test]
509    fn queue_status_roundtrip() {
510        let variants = [
511            QueueStatus::Pending,
512            QueueStatus::InFlight,
513            QueueStatus::Delivered,
514            QueueStatus::Failed,
515            QueueStatus::Bounced,
516        ];
517        for v in &variants {
518            let s = v.as_str();
519            let parsed = QueueStatus::parse(s).unwrap();
520            assert_eq!(&parsed, v, "roundtrip failed for {s}");
521        }
522    }
523
524    #[test]
525    fn queue_status_parse_unknown() {
526        assert_eq!(QueueStatus::parse("unknown"), None);
527        assert_eq!(QueueStatus::parse(""), None);
528        assert_eq!(QueueStatus::parse("PENDING"), None);
529    }
530
531    #[test]
532    fn queue_status_as_str_values() {
533        assert_eq!(QueueStatus::Pending.as_str(), "pending");
534        assert_eq!(QueueStatus::InFlight.as_str(), "inflight");
535        assert_eq!(QueueStatus::Delivered.as_str(), "delivered");
536        assert_eq!(QueueStatus::Failed.as_str(), "failed");
537        assert_eq!(QueueStatus::Bounced.as_str(), "bounced");
538    }
539
540    #[test]
541    fn queue_status_parse_case_sensitive() {
542        // parse is case-sensitive — uppercase variants are not valid
543        assert_eq!(QueueStatus::parse("Pending"), None);
544        assert_eq!(QueueStatus::parse("InFlight"), None);
545        assert_eq!(QueueStatus::parse("DELIVERED"), None);
546        assert_eq!(QueueStatus::parse("Failed"), None);
547        assert_eq!(QueueStatus::parse("Bounced"), None);
548    }
549
550    #[test]
551    fn queue_status_parse_whitespace_rejected() {
552        assert_eq!(QueueStatus::parse(" pending"), None);
553        assert_eq!(QueueStatus::parse("pending "), None);
554        assert_eq!(QueueStatus::parse("  "), None);
555    }
556
557    #[test]
558    fn queue_status_eq() {
559        assert_eq!(QueueStatus::Pending, QueueStatus::Pending);
560        assert_ne!(QueueStatus::Pending, QueueStatus::Delivered);
561        assert_ne!(QueueStatus::Failed, QueueStatus::Bounced);
562    }
563
564    #[test]
565    fn queue_status_clone() {
566        let s = QueueStatus::InFlight;
567        let c = s.clone();
568        assert_eq!(s, c);
569    }
570
571    #[test]
572    fn queued_message_clone_preserves_fields() {
573        let msg = QueuedMessage {
574            id: 42,
575            sender: "s@example.com".into(),
576            recipient: "r@remote.com".into(),
577            domain: "remote.com".into(),
578            message_data: vec![1, 2, 3],
579            status: QueueStatus::Pending,
580            attempts: 3,
581            max_attempts: 8,
582            next_retry: 1_700_000_000,
583            last_error: Some("temporary failure".into()),
584            message_id: Some("msg-id-123".into()),
585            created_at: 1_699_000_000,
586            updated_at: 1_699_500_000,
587            is_forwarded: true,
588        };
589        let cloned = msg.clone();
590        assert_eq!(cloned.id, 42);
591        assert_eq!(cloned.sender, "s@example.com");
592        assert_eq!(cloned.recipient, "r@remote.com");
593        assert_eq!(cloned.domain, "remote.com");
594        assert_eq!(cloned.message_data, vec![1, 2, 3]);
595        assert_eq!(cloned.attempts, 3);
596        assert_eq!(cloned.max_attempts, 8);
597        assert_eq!(cloned.next_retry, 1_700_000_000);
598        assert_eq!(cloned.last_error, Some("temporary failure".into()));
599        assert_eq!(cloned.message_id, Some("msg-id-123".into()));
600        assert!(cloned.is_forwarded);
601    }
602
603    #[test]
604    fn queued_message_no_last_error() {
605        let msg = QueuedMessage {
606            id: 1,
607            sender: "s@example.com".into(),
608            recipient: "r@remote.com".into(),
609            domain: "remote.com".into(),
610            message_data: vec![],
611            status: QueueStatus::Pending,
612            attempts: 0,
613            max_attempts: 8,
614            next_retry: 0,
615            last_error: None,
616            message_id: None,
617            created_at: 0,
618            updated_at: 0,
619            is_forwarded: false,
620        };
621        assert!(msg.last_error.is_none());
622        assert!(msg.message_id.is_none());
623    }
624}
mailrs_outbound_queue/queue.rs

mailrs_outbound_queue/
queue.rs
